Compare commits

..

3 Commits

Author SHA1 Message Date
025ac7b810 fix: endpoints result 2025-09-23 21:29:18 +02:00
a9f56c1279 fix: tasks endpoint response 2025-09-23 21:03:13 +02:00
594b5e3af6 feat: implement complete JSON serialization system with response classes
- Add abstract JsonSerializable base class with toJson/fromJson methods
- Create comprehensive response classes for complete JSON document handling:
  * ClusterMembersResponse for cluster member data
  * TaskStatusResponse and TaskControlResponse for task operations
  * NodeStatusResponse, NodeEndpointsResponse, NodeOperationResponse for node data
- Implement concrete serializable classes for all data types:
  * NodeInfoSerializable, TaskInfoSerializable, SystemInfoSerializable
  * TaskSummarySerializable, EndpointInfoSerializable
- Refactor all service classes to use new serialization system
- Reduce service method complexity from 20-30 lines to 2-3 lines
- Eliminate manual JsonDocument creation and field mapping
- Ensure type safety and compile-time validation
- Maintain backward compatibility while improving maintainability

Breaking change: Service classes now use response objects instead of manual JSON creation
2025-09-22 21:55:25 +02:00
43 changed files with 1111 additions and 1625 deletions

View File

@@ -9,8 +9,6 @@ SPORE is a cluster engine for ESP8266 microcontrollers that provides automatic n
- [Features](#features) - [Features](#features)
- [Supported Hardware](#supported-hardware) - [Supported Hardware](#supported-hardware)
- [Architecture](#architecture) - [Architecture](#architecture)
- [Cluster Broadcast](#cluster-broadcast)
- [Streaming API](#streaming-api)
- [API Reference](#api-reference) - [API Reference](#api-reference)
- [Configuration](#configuration) - [Configuration](#configuration)
- [Development](#development) - [Development](#development)
@@ -28,8 +26,6 @@ SPORE is a cluster engine for ESP8266 microcontrollers that provides automatic n
- **Service Registry**: Dynamic API endpoint discovery and registration - **Service Registry**: Dynamic API endpoint discovery and registration
- **Health Monitoring**: Real-time node status tracking with resource monitoring - **Health Monitoring**: Real-time node status tracking with resource monitoring
- **Event System**: Local and cluster-wide event publishing/subscription - **Event System**: Local and cluster-wide event publishing/subscription
- **Cluster Broadcast**: Centralized UDP broadcast of events (CLUSTER_EVENT)
- **Streaming API**: WebSocket bridge for real-time event send/receive
- **Over-The-Air Updates**: Seamless firmware updates across the cluster - **Over-The-Air Updates**: Seamless firmware updates across the cluster
- **REST API**: HTTP-based cluster management and monitoring - **REST API**: HTTP-based cluster management and monitoring
- **Capability Discovery**: Automatic API endpoint and service capability detection - **Capability Discovery**: Automatic API endpoint and service capability detection
@@ -107,52 +103,6 @@ void setup() {
**Examples:** See [`examples/base/`](./examples/base/) for basic usage and [`examples/relay/`](./examples/relay/) for custom service integration. **Examples:** See [`examples/base/`](./examples/base/) for basic usage and [`examples/relay/`](./examples/relay/) for custom service integration.
## Cluster Broadcast
Broadcast an event to all peers using the centralized core broadcaster. Services never touch UDP directly; instead they fire a local event that the core transmits as a `CLUSTER_EVENT`.
Usage:
```cpp
// 1) Apply locally via the same event your service already handles
JsonDocument payload;
payload["pattern"] = "rainbow_cycle";
payload["brightness"] = 100;
String payloadStr; serializeJson(payload, payloadStr);
ctx.fire("api/neopattern", &payloadStr);
// 2) Broadcast to peers via the core
JsonDocument envelope;
envelope["event"] = "api/neopattern";
envelope["data"] = payloadStr; // JSON string
String eventJson; serializeJson(envelope, eventJson);
ctx.fire("cluster/broadcast", &eventJson);
```
Notes:
- The core sends subnet-directed broadcasts (e.g., 192.168.1.255) for reliability.
- Peers receive `CLUSTER_EVENT` and forward to local subscribers with `ctx.fire(event, data)`.
- `data` can be a JSON string or nested JSON; receivers handle both.
📖 See the dedicated guide: [`docs/ClusterBroadcast.md`](./docs/ClusterBroadcast.md)
## Streaming API
Real-time event bridge available at `/ws` using WebSocket.
- Send JSON `{ event, payload }` to dispatch events via `ctx.fire`.
- Receive all local events as `{ event, payload }`.
Examples:
```json
{ "event": "api/neopattern/color", "payload": { "color": "#FF0000", "brightness": 128 } }
```
```json
{ "event": "cluster/broadcast", "payload": { "event": "api/neopattern/color", "data": { "color": "#00FF00" } } }
```
📖 See the dedicated guide: [`docs/StreamingAPI.md`](./docs/StreamingAPI.md)
## API Reference ## API Reference
The system provides a comprehensive RESTful API for monitoring and controlling the embedded device. All endpoints return JSON responses and support standard HTTP status codes. The system provides a comprehensive RESTful API for monitoring and controlling the embedded device. All endpoints return JSON responses and support standard HTTP status codes.

View File

@@ -15,18 +15,12 @@ The SPORE system provides a comprehensive RESTful API for monitoring and control
| Endpoint | Method | Description | Response | | Endpoint | Method | Description | Response |
|----------|--------|-------------|----------| |----------|--------|-------------|----------|
| `/api/node/status` | GET | System resource information | System metrics | | `/api/node/status` | GET | System resource information and API endpoint registry | System metrics and API catalog |
| `/api/node/endpoints` | GET | API endpoints and parameters | Detailed endpoint specifications | | `/api/node/endpoints` | GET | API endpoints and parameters | Detailed endpoint specifications |
| `/api/cluster/members` | GET | Cluster membership and node health information | Cluster topology and health status | | `/api/cluster/members` | GET | Cluster membership and node health information | Cluster topology and health status |
| `/api/node/update` | POST | Handle firmware updates via OTA | Update progress and status | | `/api/node/update` | POST | Handle firmware updates via OTA | Update progress and status |
| `/api/node/restart` | POST | Trigger system restart | Restart confirmation | | `/api/node/restart` | POST | Trigger system restart | Restart confirmation |
### Monitoring API
| Endpoint | Method | Description | Response |
|----------|--------|-------------|----------|
| `/api/monitoring/resources` | GET | CPU, memory, filesystem, and uptime | System resource metrics |
### Network Management API ### Network Management API
| Endpoint | Method | Description | Response | | Endpoint | Method | Description | Response |
@@ -146,7 +140,7 @@ Controls the execution state of individual tasks. Supports enabling, disabling,
#### GET /api/node/status #### GET /api/node/status
Returns comprehensive system resource information including memory usage and chip details. For a list of available API endpoints, use `/api/node/endpoints`. Returns comprehensive system resource information including memory usage, chip details, and a registry of all available API endpoints.
**Response Fields:** **Response Fields:**
- `freeHeap`: Available RAM in bytes - `freeHeap`: Available RAM in bytes
@@ -174,7 +168,7 @@ Returns comprehensive system resource information including memory usage and chi
#### GET /api/node/endpoints #### GET /api/node/endpoints
Returns detailed information about all available API endpoints, including their parameters, types, and validation rules. Methods are returned as strings (e.g., "GET", "POST"). Returns detailed information about all available API endpoints, including their parameters, types, and validation rules.
**Response Fields:** **Response Fields:**
- `endpoints[]`: Array of endpoint capability objects - `endpoints[]`: Array of endpoint capability objects
@@ -242,54 +236,6 @@ Initiates an over-the-air firmware update. The firmware file should be uploaded
Triggers a system restart. The response will be sent before the restart occurs. Triggers a system restart. The response will be sent before the restart occurs.
### Monitoring
#### GET /api/monitoring/resources
Returns real-time system resource metrics.
Response Fields:
- `cpu.current_usage`: Current CPU usage percent
- `cpu.average_usage`: Average CPU usage percent
- `cpu.max_usage`: Max observed CPU usage
- `cpu.min_usage`: Min observed CPU usage
- `cpu.measurement_count`: Number of measurements
- `cpu.is_measuring`: Whether measurement is active
- `memory.free_heap`: Free heap bytes
- `memory.total_heap`: Total heap bytes (approximate)
- `memory.heap_fragmentation`: Fragmentation percent (0 on ESP8266)
- `filesystem.total_bytes`: LittleFS total bytes
- `filesystem.used_bytes`: Used bytes
- `filesystem.free_bytes`: Free bytes
- `filesystem.usage_percent`: Usage percent
- `system.uptime_ms`: Uptime in milliseconds
Example Response:
```json
{
"cpu": {
"current_usage": 3.5,
"average_usage": 2.1,
"max_usage": 15.2,
"min_usage": 0.0,
"measurement_count": 120,
"is_measuring": true
},
"memory": {
"free_heap": 48748,
"total_heap": 81920,
"heap_fragmentation": 0
},
"filesystem": {
"total_bytes": 65536,
"used_bytes": 10240,
"free_bytes": 55296,
"usage_percent": 15.6
},
"system": {
"uptime_ms": 123456
}
}
```
### Network Management ### Network Management
#### GET /api/network/status #### GET /api/network/status

View File

@@ -25,9 +25,9 @@ The system architecture consists of several key components working together:
- **Service Registry**: Track available services across the cluster - **Service Registry**: Track available services across the cluster
### Task Scheduler ### Task Scheduler
- **Cooperative Multitasking**: Background task management system (`TaskManager`) - **Cooperative Multitasking**: Background task management system
- **Task Lifecycle Management**: Enable/disable tasks and set intervals at runtime - **Task Lifecycle Management**: Automatic task execution and monitoring
- **Execution Model**: Tasks run in `Spore::loop()` when their interval elapses - **Resource Optimization**: Efficient task scheduling and execution
### Node Context ### Node Context
- **Central Context**: Shared resources and configuration - **Central Context**: Shared resources and configuration
@@ -40,75 +40,27 @@ The cluster uses a UDP-based discovery protocol for automatic node detection:
### Discovery Process ### Discovery Process
1. **Discovery Broadcast**: Nodes periodically send UDP packets on port `udp_port` (default 4210) 1. **Discovery Broadcast**: Nodes periodically send UDP packets on port 4210
2. **Response Handling**: Nodes respond with `CLUSTER_RESPONSE:<hostname>` 2. **Response Handling**: Nodes respond with their hostname and IP address
3. **Member Management**: Discovered nodes are added/updated in the cluster 3. **Member Management**: Discovered nodes are automatically added to the cluster
4. **Node Info via UDP**: Heartbeat triggers peers to send `CLUSTER_NODE_INFO:<hostname>:<json>` 4. **Health Monitoring**: Continuous status checking via HTTP API calls
### Protocol Details ### Protocol Details
- **UDP Port**: 4210 (configurable via `Config.udp_port`) - **UDP Port**: 4210 (configurable)
- **Discovery Message**: `CLUSTER_DISCOVERY` - **Discovery Message**: `CLUSTER_DISCOVERY`
- **Response Message**: `CLUSTER_RESPONSE` - **Response Message**: `CLUSTER_RESPONSE`
- **Heartbeat Message**: `CLUSTER_HEARTBEAT`
- **Node Info Message**: `CLUSTER_NODE_INFO:<hostname>:<json>`
- **Broadcast Address**: 255.255.255.255 - **Broadcast Address**: 255.255.255.255
- **Discovery Interval**: `Config.discovery_interval_ms` (default 1000 ms) - **Discovery Interval**: 1 second (configurable)
- **Listen Interval**: `Config.cluster_listen_interval_ms` (default 10 ms) - **Listen Interval**: 100ms (configurable)
- **Heartbeat Interval**: `Config.heartbeat_interval_ms` (default 5000 ms)
### Message Formats
- **Discovery**: `CLUSTER_DISCOVERY`
- Sender: any node, broadcast to 255.255.255.255:`udp_port`
- Purpose: announce presence and solicit peer identification
- **Response**: `CLUSTER_RESPONSE:<hostname>`
- Sender: node receiving a discovery; unicast to requester IP
- Purpose: provide hostname so requester can register/update member
- **Heartbeat**: `CLUSTER_HEARTBEAT:<hostname>`
- Sender: each node, broadcast to 255.255.255.255:`udp_port` on interval
- Purpose: prompt peers to reply with their node info and keep liveness
- **Node Info**: `CLUSTER_NODE_INFO:<hostname>:<json>`
- Sender: node receiving a heartbeat; unicast to heartbeat sender IP
- JSON fields: freeHeap, chipId, sdkVersion, cpuFreqMHz, flashChipSize, optional labels
### Discovery Flow
1. **Sender broadcasts** `CLUSTER_DISCOVERY`
2. **Each receiver responds** with `CLUSTER_RESPONSE:<hostname>` to the sender IP
3. **Sender registers/updates** the node using hostname and source IP
### Heartbeat Flow
1. **A node broadcasts** `CLUSTER_HEARTBEAT:<hostname>`
2. **Each receiver replies** with `CLUSTER_NODE_INFO:<hostname>:<json>` to the heartbeat sender IP
3. **The sender**:
- Ensures the node exists or creates it with `hostname` and sender IP
- Parses JSON and updates resources, labels, `status = ACTIVE`, `lastSeen = now`
- Sets `latency = now - lastHeartbeatSentAt` (per-node, measured at heartbeat origin)
### Listener Behavior
The `cluster_listen` task parses one UDP packet per run and dispatches by prefix to:
- **Discovery** → send `CLUSTER_RESPONSE`
- **Heartbeat** → send `CLUSTER_NODE_INFO` JSON
- **Response** → add/update node using provided hostname and source IP
- **Node Info** → update resources/status/labels and record latency
### Timing and Intervals
- **UDP Port**: `Config.udp_port` (default 4210)
- **Discovery Interval**: `Config.discovery_interval_ms` (default 1000 ms)
- **Listen Interval**: `Config.cluster_listen_interval_ms` (default 10 ms)
- **Heartbeat Interval**: `Config.heartbeat_interval_ms` (default 5000 ms)
### Node Status Categories ### Node Status Categories
Nodes are automatically categorized by their activity: Nodes are automatically categorized by their activity:
- **ACTIVE**: lastSeen < `node_inactive_threshold_ms` (default 10s) - **ACTIVE**: Responding within 10 seconds
- **INACTIVE**: < `node_dead_threshold_ms` (default 120s) - **INACTIVE**: No response for 10-60 seconds
- **DEAD**: `node_dead_threshold_ms` - **DEAD**: No response for over 60 seconds
## Task Scheduling System ## Task Scheduling System
@@ -116,14 +68,14 @@ The system runs several background tasks at different intervals:
### Core System Tasks ### Core System Tasks
| Task | Interval (default) | Purpose | | Task | Interval | Purpose |
|------|--------------------|---------| |------|----------|---------|
| `cluster_discovery` | 1000 ms | Send UDP discovery packets | | **Discovery Send** | 1 second | Send UDP discovery packets |
| `cluster_listen` | 10 ms | Listen for discovery/heartbeat/node-info | | **Discovery Listen** | 100ms | Listen for discovery responses |
| `status_update` | 1000 ms | Update node status categories, purge dead | | **Status Updates** | 1 second | Monitor cluster member health |
| `heartbeat` | 5000 ms | Broadcast heartbeat and update local resources | | **Heartbeat** | 2 seconds | Maintain cluster connectivity |
| `cluster_update_members_info` | 10000 ms | Reserved; no-op (info via UDP) | | **Member Info** | 10 seconds | Update detailed node information |
| `print_members` | 5000 ms | Log current member list | | **Debug Output** | 5 seconds | Print cluster status |
### Task Management Features ### Task Management Features
@@ -160,7 +112,10 @@ ctx.fire("cluster_updated", &clusterData);
### Available Events ### Available Events
- **`node_discovered`**: New node added or local node refreshed - **`node_discovered`**: New node added to cluster
- **`cluster_updated`**: Cluster membership changed
- **`resource_update`**: Node resources updated
- **`health_check`**: Node health status changed
## Resource Monitoring ## Resource Monitoring
@@ -200,8 +155,10 @@ The system includes automatic WiFi fallback for robust operation:
### Configuration ### Configuration
- **Hostname**: Derived from MAC (`esp-<mac>`) and assigned to `ctx.hostname` - **SSID Format**: `SPORE_<MAC_LAST_4>`
- **AP Mode**: If STA connection fails, device switches to AP mode with configured SSID/password - **Password**: Configurable fallback password
- **IP Range**: 192.168.4.x subnet
- **Gateway**: 192.168.4.1
## Cluster Topology ## Cluster Topology
@@ -213,30 +170,32 @@ The system includes automatic WiFi fallback for robust operation:
### Network Architecture ### Network Architecture
- UDP broadcast-based discovery and heartbeats on local subnet - **Mesh-like Structure**: Nodes can communicate with each other
- Optional HTTP polling (disabled by default; node info exchanged via UDP) - **Dynamic Routing**: Automatic path discovery between nodes
- **Load Distribution**: Tasks distributed across available nodes
- **Fault Tolerance**: Automatic failover and recovery
## Data Flow ## Data Flow
### Node Discovery ### Node Discovery
1. **UDP Broadcast**: Nodes broadcast discovery packets on port 4210 1. **UDP Broadcast**: Nodes broadcast discovery packets on port 4210
2. **UDP Response**: Receiving nodes respond with hostname 2. **UDP Response**: Receiving nodes responds with hostname
3. **Registration**: Discovered nodes are added to local cluster member list 3. **Registration**: Discovered nodes are added to local cluster member list
### Health Monitoring ### Health Monitoring
1. **Periodic Checks**: Cluster manager updates node status categories 1. **Periodic Checks**: Cluster manager polls member nodes every 1 second
2. **Status Collection**: Each node updates resources via UDP node-info messages 2. **Status Collection**: Each node returns resource usage and health metrics
### Task Management ### Task Management
1. **Scheduling**: `TaskManager` executes registered tasks at configured intervals 1. **Scheduling**: TaskScheduler executes registered tasks at configured intervals
2. **Execution**: Tasks run cooperatively in the main loop without preemption 2. **Execution**: Tasks run cooperatively, yielding control to other tasks
3. **Monitoring**: Task status is exposed via REST (`/api/tasks/status`) 3. **Monitoring**: Task status and results are exposed via REST API endpoints
## Performance Characteristics ## Performance Characteristics
### Memory Usage ### Memory Usage
- **Base System**: ~15-20KB RAM (device dependent) - **Base System**: ~15-20KB RAM
- **Per Task**: ~100-200 bytes per task - **Per Task**: ~100-200 bytes per task
- **Cluster Members**: ~50-100 bytes per member - **Cluster Members**: ~50-100 bytes per member
- **API Endpoints**: ~20-30 bytes per endpoint - **API Endpoints**: ~20-30 bytes per endpoint
@@ -260,7 +219,7 @@ The system includes automatic WiFi fallback for robust operation:
### Current Implementation ### Current Implementation
- **Network Access**: Local network only (no internet exposure) - **Network Access**: Local network only (no internet exposure)
- **Authentication**: None currently implemented; LAN-only access assumed - **Authentication**: None currently implemented
- **Data Validation**: Basic input validation - **Data Validation**: Basic input validation
- **Resource Limits**: Memory and processing constraints - **Resource Limits**: Memory and processing constraints

View File

@@ -1,91 +0,0 @@
## Cluster Broadcast (CLUSTER_EVENT)
### Overview
Spore supports cluster-wide event broadcasting via UDP. Services publish a local event, and the core broadcasts it to peers as a `CLUSTER_EVENT`. Peers receive the event and forward it to local subscribers through the internal event bus.
- **Local trigger**: `ctx.fire("cluster/broadcast", eventJson)`
- **UDP message**: `CLUSTER_EVENT:{json}`
- **Receiver action**: Parses `{json}` and calls `ctx.fire(event, data)`
This centralizes network broadcast in core, so services never touch UDP directly.
### Message format
- UDP payload prefix: `CLUSTER_EVENT:`
- JSON body:
```json
{
"event": "<event-name>",
"data": "<json-string>" // or an inline JSON object/array
}
```
Notes:
- The receiver accepts `data` as either a JSON string or a nested JSON object/array. Nested JSON is serialized back to a string before firing the local event.
- Keep payloads small (UDP, default buffer 512 bytes).
### Core responsibilities
- `ClusterManager` registers a centralized handler:
- Subscribes to `cluster/broadcast` to send the provided event JSON over UDP broadcast.
- Listens for incoming UDP `CLUSTER_EVENT` messages and forwards them to local subscribers via `ctx.fire(event, data)`.
- Broadcast target uses subnet-directed broadcast (e.g., `192.168.1.255`) for better reliability. Both nodes must share the same `udp_port`.
### Service responsibilities
Services send and receive events using the local event bus.
1) Subscribe to an event name and apply state from `data`:
```cpp
ctx.on("api/neopattern", [this](void* dataPtr) {
String* jsonStr = static_cast<String*>(dataPtr);
if (!jsonStr) return;
JsonDocument doc;
if (deserializeJson(doc, *jsonStr)) return;
JsonObject obj = doc.as<JsonObject>();
// Parse and apply fields from obj
});
```
2) Build a control payload and update locally via the same event:
```cpp
JsonDocument payload;
payload["pattern"] = "rainbow_cycle"; // example
payload["brightness"] = 100;
String payloadStr; serializeJson(payload, payloadStr);
ctx.fire("api/neopattern", &payloadStr);
``;
3) Broadcast to peers by delegating to core:
```cpp
JsonDocument envelope;
envelope["event"] = "api/neopattern";
envelope["data"] = payloadStr; // JSON string
String eventJson; serializeJson(envelope, eventJson);
ctx.fire("cluster/broadcast", &eventJson);
```
With this flow, services have a single codepath for applying state (the event handler). Broadcasting simply reuses the same payload.
### Logging
- Core logs source IP, payload length, and event name for received `CLUSTER_EVENT`s.
- Services can log when submitting `cluster/broadcast` and when applying control events.
### Networking considerations
- Ensure all nodes:
- Listen on the same `udp_port`.
- Are in the same subnet (for subnet-directed broadcast).
- Some networks may block global broadcast (`255.255.255.255`). Subnet-directed broadcast is used by default.
### Troubleshooting
- If peers do not react:
- Confirm logs show `CLUSTER_EVENT raw from <ip>` on the receiver.
- Verify UDP port alignment and WiFi connection/subnet.
- Check payload size (<512 bytes by default) and JSON validity.
- Ensure the service subscribed to the correct `event` name and handles `data`.

View File

@@ -20,99 +20,57 @@
``` ```
spore/ spore/
├── src/ # Source code (framework under src/spore) ├── src/ # Source code
── spore/ ── main.cpp # Main application entry point
├── Spore.cpp # Framework lifecycle (setup/begin/loop) ├── ApiServer.cpp # HTTP API server implementation
├── core/ # Core components ├── ClusterManager.cpp # Cluster management logic
│ ├── ApiServer.cpp # HTTP API server implementation ├── NetworkManager.cpp # WiFi and network handling
│ ├── ClusterManager.cpp # Cluster management logic │ ├── TaskManager.cpp # Background task management
│ ├── NetworkManager.cpp # WiFi and network handling └── NodeContext.cpp # Central context and events
│ │ ├── TaskManager.cpp # Background task management
│ │ └── NodeContext.cpp # Central context and events
│ ├── services/ # Built-in services
│ │ ├── NodeService.cpp
│ │ ├── NetworkService.cpp
│ │ ├── ClusterService.cpp
│ │ ├── TaskService.cpp
│ │ ├── StaticFileService.cpp
│ │ └── MonitoringService.cpp
│ └── types/ # Shared types
├── include/ # Header files ├── include/ # Header files
├── examples/ # Example apps per env (base, relay, neopattern) ├── lib/ # Library files
├── docs/ # Documentation ├── docs/ # Documentation
├── api/ # OpenAPI specification ├── api/ # OpenAPI specification
├── platformio.ini # PlatformIO configuration ├── examples/ # Example code
── ctl.sh # Build and deployment scripts ── test/ # Test files
├── platformio.ini # PlatformIO configuration
└── ctl.sh # Build and deployment scripts
``` ```
## PlatformIO Configuration ## PlatformIO Configuration
### Framework and Board ### Framework and Board
The project uses PlatformIO with the following configuration (excerpt): The project uses PlatformIO with the following configuration:
```ini ```ini
[platformio] [env:esp01_1m]
default_envs = base
src_dir = .
data_dir = ${PROJECT_DIR}/examples/${PIOENV}/data
[common]
monitor_speed = 115200
lib_deps =
esp32async/ESPAsyncWebServer@^3.8.0
bblanchon/ArduinoJson@^7.4.2
[env:base]
platform = platformio/espressif8266@^4.2.1 platform = platformio/espressif8266@^4.2.1
board = esp01_1m board = esp01_1m
framework = arduino framework = arduino
upload_speed = 115200 upload_speed = 115200
monitor_speed = 115200 flash_mode = dout
board_build.f_cpu = 80000000L
board_build.flash_mode = qio
board_build.filesystem = littlefs
; note: somehow partition table is not working, so we need to use the ldscript
board_build.ldscript = eagle.flash.1m64.ld
lib_deps = ${common.lib_deps}
build_src_filter =
+<examples/base/*.cpp>
+<src/spore/*.cpp>
+<src/spore/core/*.cpp>
+<src/spore/services/*.cpp>
+<src/spore/types/*.cpp>
+<src/spore/util/*.cpp>
+<src/internal/*.cpp>
[env:d1_mini]
platform = platformio/espressif8266@^4.2.1
board = d1_mini
framework = arduino
upload_speed = 115200
monitor_speed = 115200
board_build.filesystem = littlefs
board_build.flash_mode = dio ; D1 Mini uses DIO on 4 Mbit flash
board_build.flash_size = 4M
board_build.ldscript = eagle.flash.4m1m.ld
lib_deps = ${common.lib_deps}
build_src_filter =
+<examples/base/*.cpp>
+<src/spore/*.cpp>
+<src/spore/core/*.cpp>
+<src/spore/services/*.cpp>
+<src/spore/types/*.cpp>
+<src/spore/util/*.cpp>
+<src/internal/*.cpp>
``` ```
### Key Configuration Details
- **Framework**: Arduino
- **Board**: ESP-01 with 1MB flash
- **Upload Speed**: 115200 baud
- **Flash Mode**: DOUT (required for ESP-01S)
- **Build Type**: Release (optimized for production)
### Dependencies ### Dependencies
The project requires the following libraries (resolved via PlatformIO): The project requires the following libraries:
```ini ```ini
lib_deps = lib_deps =
esp32async/ESPAsyncWebServer@^3.8.0 esp32async/ESPAsyncWebServer@^3.8.0
bblanchon/ArduinoJson@^7.4.2 bblanchon/ArduinoJson@^7.4.2
arkhipenko/TaskScheduler@^3.8.5
ESP8266HTTPClient@1.2
ESP8266WiFi@1.0
``` ```
### Filesystem, Linker Scripts, and Flash Layout ### Filesystem, Linker Scripts, and Flash Layout
@@ -145,6 +103,7 @@ Notes:
- If you need a different FS size, select an appropriate ldscript variant and keep `board_build.filesystem = littlefs`. - If you need a different FS size, select an appropriate ldscript variant and keep `board_build.filesystem = littlefs`.
- On ESP8266, custom partition CSVs are not used for layout; the linker script defines the flash map. This project removed prior `board_build.partitions` usage in favor of explicit `board_build.ldscript` entries per environment. - On ESP8266, custom partition CSVs are not used for layout; the linker script defines the flash map. This project removed prior `board_build.partitions` usage in favor of explicit `board_build.ldscript` entries per environment.
## Building ## Building
### Basic Build Commands ### Basic Build Commands
@@ -349,7 +308,7 @@ export API_NODE=192.168.1.100
Key configuration files: Key configuration files:
- **`platformio.ini`**: Build and upload configuration - **`platformio.ini`**: Build and upload configuration
- **`src/spore/types/Config.cpp`**: Default runtime configuration - **`src/Config.cpp`**: Application configuration
- **`.env`**: Environment variables - **`.env`**: Environment variables
- **`ctl.sh`**: Build and deployment scripts - **`ctl.sh`**: Build and deployment scripts

View File

@@ -1,79 +0,0 @@
# Monitoring Service
Exposes system resource metrics via HTTP for observability.
## Overview
- **Service name**: `MonitoringService`
- **Endpoint**: `GET /api/monitoring/resources`
- **Metrics**: CPU usage, memory, filesystem, uptime
## Endpoint
### GET /api/monitoring/resources
Returns real-time system resource metrics.
Response fields:
- `cpu.current_usage`: Current CPU usage percent
- `cpu.average_usage`: Average CPU usage percent
- `cpu.max_usage`: Max observed CPU usage
- `cpu.min_usage`: Min observed CPU usage
- `cpu.measurement_count`: Number of measurements
- `cpu.is_measuring`: Whether measurement is active
- `memory.free_heap`: Free heap bytes
- `memory.total_heap`: Total heap bytes (approximate)
- `memory.min_free_heap`: Minimum free heap (0 on ESP8266)
- `memory.max_alloc_heap`: Max allocatable heap (0 on ESP8266)
- `memory.heap_fragmentation`: Fragmentation percent (0 on ESP8266)
- `filesystem.total_bytes`: LittleFS total bytes
- `filesystem.used_bytes`: Used bytes
- `filesystem.free_bytes`: Free bytes
- `filesystem.usage_percent`: Usage percent
- `system.uptime_ms`: Uptime in milliseconds
- `system.uptime_seconds`: Uptime in seconds
- `system.uptime_formatted`: Human-readable uptime
Example:
```json
{
"cpu": {
"current_usage": 3.5,
"average_usage": 2.1,
"max_usage": 15.2,
"min_usage": 0.0,
"measurement_count": 120,
"is_measuring": true
},
"memory": {
"free_heap": 48748,
"total_heap": 81920,
"min_free_heap": 0,
"max_alloc_heap": 0,
"heap_fragmentation": 0,
"heap_usage_percent": 40.4
},
"filesystem": {
"total_bytes": 65536,
"used_bytes": 10240,
"free_bytes": 55296,
"usage_percent": 15.6
},
"system": {
"uptime_ms": 123456,
"uptime_seconds": 123,
"uptime_formatted": "0h 2m 3s"
}
}
```
## Implementation Notes
- `MonitoringService` reads from `CpuUsage` and ESP8266 SDK APIs.
- Filesystem metrics are gathered from LittleFS.
- CPU measurement is bracketed by `Spore::loop()` calling `cpuUsage.startMeasurement()` and `cpuUsage.endMeasurement()`.
## Troubleshooting
- If `filesystem.total_bytes` is zero, ensure LittleFS is enabled in `platformio.ini` and an FS image is uploaded.
- CPU usage values remain zero until the main loop runs and CPU measurement is started.

View File

@@ -15,8 +15,15 @@ Complete API reference with detailed endpoint documentation, examples, and integ
- Task management workflows - Task management workflows
- Cluster monitoring examples - Cluster monitoring examples
### 📖 [MonitoringService.md](./MonitoringService.md) ### 📖 [TaskManager.md](./TaskManager.md)
System resource monitoring API for CPU, memory, filesystem, and uptime. Comprehensive guide to the TaskManager system for background task management.
**Includes:**
- Basic usage examples
- Advanced binding techniques
- Task status monitoring
- API integration details
- Performance considerations
### 📖 [TaskManagement.md](./TaskManagement.md) ### 📖 [TaskManagement.md](./TaskManagement.md)
Complete guide to the task management system with examples and best practices. Complete guide to the task management system with examples and best practices.

View File

@@ -1,80 +0,0 @@
## Streaming API (WebSocket)
### Overview
The streaming API exposes an event-driven WebSocket at `/ws`. It bridges between external clients and the internal event bus:
- Incoming WebSocket JSON `{ event, payload }``ctx.fire(event, payload)`
- Local events → broadcasted to all connected WebSocket clients as `{ event, payload }`
This allows real-time control and observation of the system without polling.
### URL
- `ws://<device-ip>/ws`
### Message Format
- Client → Device
```json
{
"event": "<event-name>",
"payload": "<json-string>" | { /* inline JSON */ }
}
```
- Device → Client
```json
{
"event": "<event-name>",
"payload": "<json-string>"
}
```
Notes:
- The device accepts `payload` as a string or a JSON object/array. Objects are serialized into a string before dispatching to local subscribers to keep a consistent downstream contract.
- A minimal ack `{ "ok": true }` is sent after a valid inbound message.
#### Echo suppression (origin tagging)
- To prevent the sender from receiving an immediate echo of its own message, the server injects a private field into JSON payloads:
- `_origin: "ws:<clientId>"`
- When re-broadcasting local events to WebSocket clients, the server:
- Strips the `_origin` field from the outgoing payload
- Skips the originating `clientId` so only other clients receive the message
- If a payload is not valid JSON (plain string), no origin tag is injected and the message may be echoed
### Event Bus Integration
- The WebSocket registers an `onAny` subscriber to `NodeContext` so that all local events are mirrored to clients.
- Services should subscribe to specific events via `ctx.on("<name>", ...)`.
### Examples
1) Set a solid color on NeoPattern:
```json
{
"event": "api/neopattern/color",
"payload": { "color": "#FF0000", "brightness": 128 }
}
```
2) Broadcast a cluster event (delegated to core):
```json
{
"event": "cluster/broadcast",
"payload": {
"event": "api/neopattern/color",
"data": { "color": "#00FF00", "brightness": 128 }
}
}
```
### Reference Implementation
- WebSocket setup and bridging are implemented in `ApiServer`.
- Global event subscription uses `NodeContext::onAny`.
Related docs:
- [`ClusterBroadcast.md`](./ClusterBroadcast.md) — centralized UDP broadcasting and CLUSTER_EVENT format

View File

@@ -319,18 +319,18 @@ curl -X POST http://192.168.1.100/api/tasks/control \
### Before (with wrapper functions): ### Before (with wrapper functions):
```cpp ```cpp
void discoverySendTask() { cluster.sendDiscovery(); } void discoverySendTask() { cluster.sendDiscovery(); }
void clusterListenTask() { cluster.listen(); } void discoveryListenTask() { cluster.listenForDiscovery(); }
taskManager.registerTask("discovery_send", interval, discoverySendTask); taskManager.registerTask("discovery_send", interval, discoverySendTask);
taskManager.registerTask("cluster_listen", interval, clusterListenTask); taskManager.registerTask("discovery_listen", interval, discoveryListenTask);
``` ```
### After (with std::bind): ### After (with std::bind):
```cpp ```cpp
taskManager.registerTask("discovery_send", interval, taskManager.registerTask("discovery_send", interval,
std::bind(&ClusterManager::sendDiscovery, &cluster)); std::bind(&ClusterManager::sendDiscovery, &cluster));
taskManager.registerTask("cluster_listen", interval, taskManager.registerTask("discovery_listen", interval,
std::bind(&ClusterManager::listen, &cluster)); std::bind(&ClusterManager::listenForDiscovery, &cluster));
``` ```
## Compatibility ## Compatibility

View File

@@ -1,13 +1,10 @@
#include "NeoPatternService.h" #include "NeoPatternService.h"
#include "spore/core/ApiServer.h" #include "spore/core/ApiServer.h"
#include "spore/util/Logging.h" #include "spore/util/Logging.h"
#include "spore/internal/Globals.h"
#include <ArduinoJson.h> #include <ArduinoJson.h>
#include <ESP8266WiFi.h>
NeoPatternService::NeoPatternService(NodeContext& ctx, TaskManager& taskMgr, const NeoPixelConfig& config) NeoPatternService::NeoPatternService(TaskManager& taskMgr, const NeoPixelConfig& config)
: taskManager(taskMgr), : taskManager(taskMgr),
ctx(ctx),
config(config), config(config),
activePattern(NeoPatternType::RAINBOW_CYCLE), activePattern(NeoPatternType::RAINBOW_CYCLE),
direction(NeoDirection::FORWARD), direction(NeoDirection::FORWARD),
@@ -35,7 +32,6 @@ NeoPatternService::NeoPatternService(NodeContext& ctx, TaskManager& taskMgr, con
registerPatterns(); registerPatterns();
registerTasks(); registerTasks();
registerEventHandlers();
initialized = true; initialized = true;
LOG_INFO("NeoPattern", "Service initialized"); LOG_INFO("NeoPattern", "Service initialized");
@@ -68,8 +64,7 @@ void NeoPatternService::registerEndpoints(ApiServer& api) {
ParamSpec{String("brightness"), false, String("body"), String("numberRange"), {}, String("80")}, ParamSpec{String("brightness"), false, String("body"), String("numberRange"), {}, String("80")},
ParamSpec{String("total_steps"), false, String("body"), String("numberRange"), {}, String("16")}, ParamSpec{String("total_steps"), false, String("body"), String("numberRange"), {}, String("16")},
ParamSpec{String("direction"), false, String("body"), String("string"), {String("forward"), String("reverse")}}, ParamSpec{String("direction"), false, String("body"), String("string"), {String("forward"), String("reverse")}},
ParamSpec{String("interval"), false, String("body"), String("number"), {}, String("100")}, ParamSpec{String("interval"), false, String("body"), String("number"), {}, String("100")}
ParamSpec{String("broadcast"), false, String("body"), String("boolean"), {}}
}); });
// State endpoint for complex state updates // State endpoint for complex state updates
@@ -124,49 +119,61 @@ void NeoPatternService::handlePatternsRequest(AsyncWebServerRequest* request) {
void NeoPatternService::handleControlRequest(AsyncWebServerRequest* request) { void NeoPatternService::handleControlRequest(AsyncWebServerRequest* request) {
bool updated = false; bool updated = false;
bool broadcast = false;
if (request->hasParam("broadcast", true)) { if (request->hasParam("pattern", true)) {
String b = request->getParam("broadcast", true)->value(); String name = request->getParam("pattern", true)->value();
broadcast = b.equalsIgnoreCase("true") || b == "1"; if (isValidPattern(name)) {
setPatternByName(name);
updated = true;
} else {
// Invalid pattern name - could add error handling here
LOG_WARN("NeoPattern", "Invalid pattern name: " + name);
}
} }
// Build JSON payload from provided params (single source of truth) if (request->hasParam("color", true)) {
JsonDocument payload; String colorStr = request->getParam("color", true)->value();
bool any = false; uint32_t color = parseColor(colorStr);
if (request->hasParam("pattern", true)) { payload["pattern"] = request->getParam("pattern", true)->value(); any = true; } setColor(color);
if (request->hasParam("color", true)) { payload["color"] = request->getParam("color", true)->value(); any = true; }
if (request->hasParam("color2", true)) { payload["color2"] = request->getParam("color2", true)->value(); any = true; }
if (request->hasParam("brightness", true)) { payload["brightness"] = request->getParam("brightness", true)->value(); any = true; }
if (request->hasParam("total_steps", true)) { payload["total_steps"] = request->getParam("total_steps", true)->value(); any = true; }
if (request->hasParam("direction", true)) { payload["direction"] = request->getParam("direction", true)->value(); any = true; }
if (request->hasParam("interval", true)) { payload["interval"] = request->getParam("interval", true)->value(); any = true; }
String payloadStr;
serializeJson(payload, payloadStr);
// Always apply locally via event so we have a single codepath for updates
if (any) {
std::string ev = "api/neopattern";
String localData = payloadStr;
LOG_INFO("NeoPattern", String("Applying local api/neopattern via event payloadLen=") + String(payloadStr.length()));
ctx.fire(ev, &localData);
updated = true; updated = true;
} }
// Broadcast to peers if requested (delegate to core broadcast handler) if (request->hasParam("color2", true)) {
if (broadcast && any) { String colorStr = request->getParam("color2", true)->value();
JsonDocument eventDoc; uint32_t color = parseColor(colorStr);
eventDoc["event"] = "api/neopattern"; setColor2(color);
eventDoc["data"] = payloadStr; // data is JSON string updated = true;
}
String eventJson; if (request->hasParam("brightness", true)) {
serializeJson(eventDoc, eventJson); int b = request->getParam("brightness", true)->value().toInt();
if (b < 0) b = 0;
if (b > 255) b = 255;
setBrightness(static_cast<uint8_t>(b));
updated = true;
}
LOG_INFO("NeoPattern", String("Submitting cluster/broadcast for api/neopattern payloadLen=") + String(payloadStr.length())); if (request->hasParam("total_steps", true)) {
std::string ev = "cluster/broadcast"; int steps = request->getParam("total_steps", true)->value().toInt();
String eventStr = eventJson; if (steps > 0) {
ctx.fire(ev, &eventStr); setTotalSteps(static_cast<uint16_t>(steps));
updated = true;
}
}
if (request->hasParam("direction", true)) {
String dirStr = request->getParam("direction", true)->value();
NeoDirection dir = (dirStr.equalsIgnoreCase("reverse")) ? NeoDirection::REVERSE : NeoDirection::FORWARD;
setDirection(dir);
updated = true;
}
if (request->hasParam("interval", true)) {
unsigned long interval = request->getParam("interval", true)->value().toInt();
if (interval > 0) {
setUpdateInterval(interval);
updated = true;
}
} }
// Return current state // Return current state
@@ -185,139 +192,6 @@ void NeoPatternService::handleControlRequest(AsyncWebServerRequest* request) {
serializeJson(resp, json); serializeJson(resp, json);
request->send(200, "application/json", json); request->send(200, "application/json", json);
} }
void NeoPatternService::registerEventHandlers() {
ctx.on("api/neopattern", [this](void* dataPtr) {
String* jsonStr = static_cast<String*>(dataPtr);
if (!jsonStr) {
LOG_WARN("NeoPattern", "Received api/neopattern with null dataPtr");
return;
}
LOG_INFO("NeoPattern", String("Received api/neopattern event dataLen=") + String(jsonStr->length()));
JsonDocument doc;
DeserializationError err = deserializeJson(doc, *jsonStr);
if (err) {
LOG_WARN("NeoPattern", String("Failed to parse CLUSTER_EVENT data: ") + err.c_str());
return;
}
JsonObject obj = doc.as<JsonObject>();
bool applied = applyControlParams(obj);
if (applied) {
LOG_INFO("NeoPattern", "Applied control from CLUSTER_EVENT");
}
});
// Solid color event: sets all pixels to the same color
ctx.on("api/neopattern/color", [this](void* dataPtr) {
String* jsonStr = static_cast<String*>(dataPtr);
if (!jsonStr) {
LOG_WARN("NeoPattern", "Received api/neopattern/color with null dataPtr");
return;
}
JsonDocument doc;
DeserializationError err = deserializeJson(doc, *jsonStr);
if (err) {
LOG_WARN("NeoPattern", String("Failed to parse color event data: ") + err.c_str());
return;
}
JsonObject obj = doc.as<JsonObject>();
// color can be string or number
String colorStr;
if (obj["color"].is<const char*>() || obj["color"].is<String>()) {
colorStr = obj["color"].as<String>();
} else if (obj["color"].is<long>() || obj["color"].is<int>()) {
colorStr = String(obj["color"].as<long>());
} else {
LOG_WARN("NeoPattern", "api/neopattern/color missing 'color'");
return;
}
// Optional brightness
if (obj["brightness"].is<int>() || obj["brightness"].is<long>()) {
int b = obj["brightness"].as<int>();
if (b < 0) b = 0; if (b > 255) b = 255;
setBrightness(static_cast<uint8_t>(b));
}
uint32_t color = parseColor(colorStr);
setPattern(NeoPatternType::NONE);
setColor(color);
LOG_INFO("NeoPattern", String("Set solid color ") + colorStr);
});
}
bool NeoPatternService::applyControlParams(const JsonObject& obj) {
bool updated = false;
if (obj["pattern"].is<const char*>() || obj["pattern"].is<String>()) {
String name = obj["pattern"].as<String>();
if (isValidPattern(name)) {
setPatternByName(name);
updated = true;
}
}
if (obj["color"].is<const char*>() || obj["color"].is<String>() || obj["color"].is<long>() || obj["color"].is<int>()) {
String colorStr;
if (obj["color"].is<long>() || obj["color"].is<int>()) {
colorStr = String(obj["color"].as<long>());
} else {
colorStr = obj["color"].as<String>();
}
uint32_t color = parseColor(colorStr);
setColor(color);
updated = true;
}
if (obj["color2"].is<const char*>() || obj["color2"].is<String>() || obj["color2"].is<long>() || obj["color2"].is<int>()) {
String colorStr;
if (obj["color2"].is<long>() || obj["color2"].is<int>()) {
colorStr = String(obj["color2"].as<long>());
} else {
colorStr = obj["color2"].as<String>();
}
uint32_t color = parseColor(colorStr);
setColor2(color);
updated = true;
}
if (obj["brightness"].is<int>() || obj["brightness"].is<long>() || obj["brightness"].is<const char*>() || obj["brightness"].is<String>()) {
int b = 0;
if (obj["brightness"].is<int>() || obj["brightness"].is<long>()) {
b = obj["brightness"].as<int>();
} else {
b = String(obj["brightness"].as<String>()).toInt();
}
if (b < 0) {
b = 0;
}
if (b > 255) {
b = 255;
}
setBrightness(static_cast<uint8_t>(b));
updated = true;
}
if (obj["total_steps"].is<int>() || obj["total_steps"].is<long>() || obj["total_steps"].is<const char*>() || obj["total_steps"].is<String>()) {
int steps = 0;
if (obj["total_steps"].is<int>() || obj["total_steps"].is<long>()) {
steps = obj["total_steps"].as<int>();
} else {
steps = String(obj["total_steps"].as<String>()).toInt();
}
if (steps > 0) { setTotalSteps(static_cast<uint16_t>(steps)); updated = true; }
}
if (obj["direction"].is<const char*>() || obj["direction"].is<String>()) {
String dirStr = obj["direction"].as<String>();
NeoDirection dir = (dirStr.equalsIgnoreCase("reverse")) ? NeoDirection::REVERSE : NeoDirection::FORWARD;
setDirection(dir);
updated = true;
}
if (obj["interval"].is<int>() || obj["interval"].is<long>() || obj["interval"].is<const char*>() || obj["interval"].is<String>()) {
unsigned long interval = 0;
if (obj["interval"].is<int>() || obj["interval"].is<long>()) {
interval = obj["interval"].as<unsigned long>();
} else {
interval = String(obj["interval"].as<String>()).toInt();
}
if (interval > 0) { setUpdateInterval(interval); updated = true; }
}
return updated;
}
void NeoPatternService::handleStateRequest(AsyncWebServerRequest* request) { void NeoPatternService::handleStateRequest(AsyncWebServerRequest* request) {
if (request->contentType() != "application/json") { if (request->contentType() != "application/json") {

View File

@@ -1,7 +1,6 @@
#pragma once #pragma once
#include "spore/Service.h" #include "spore/Service.h"
#include "spore/core/TaskManager.h" #include "spore/core/TaskManager.h"
#include "spore/core/NodeContext.h"
#include "NeoPattern.h" #include "NeoPattern.h"
#include "NeoPatternState.h" #include "NeoPatternState.h"
#include "NeoPixelConfig.h" #include "NeoPixelConfig.h"
@@ -26,7 +25,7 @@ public:
REVERSE REVERSE
}; };
NeoPatternService(NodeContext& ctx, TaskManager& taskMgr, const NeoPixelConfig& config); NeoPatternService(TaskManager& taskMgr, const NeoPixelConfig& config);
~NeoPatternService(); ~NeoPatternService();
void registerEndpoints(ApiServer& api) override; void registerEndpoints(ApiServer& api) override;
@@ -50,8 +49,6 @@ private:
void registerTasks(); void registerTasks();
void registerPatterns(); void registerPatterns();
void update(); void update();
void registerEventHandlers();
bool applyControlParams(const JsonObject& obj);
// Pattern updaters // Pattern updaters
void updateRainbowCycle(); void updateRainbowCycle();
@@ -83,7 +80,6 @@ private:
String getPatternDescription(const String& name) const; String getPatternDescription(const String& name) const;
TaskManager& taskManager; TaskManager& taskManager;
NodeContext& ctx;
NeoPattern* neoPattern; NeoPattern* neoPattern;
NeoPixelConfig config; NeoPixelConfig config;
NeoPatternState currentState; NeoPatternState currentState;

View File

@@ -45,7 +45,7 @@ void setup() {
); );
// Create and add custom service // Create and add custom service
neoPatternService = new NeoPatternService(spore.getContext(), spore.getTaskManager(), config); neoPatternService = new NeoPatternService(spore.getTaskManager(), config);
spore.addService(neoPatternService); spore.addService(neoPatternService);
// Start the API server and complete initialization // Start the API server and complete initialization

View File

@@ -2,7 +2,6 @@
#include <Arduino.h> #include <Arduino.h>
#include <ArduinoJson.h> #include <ArduinoJson.h>
#include <ESPAsyncWebServer.h> #include <ESPAsyncWebServer.h>
#include <AsyncWebSocket.h>
#include <Updater.h> #include <Updater.h>
#include <functional> #include <functional>
#include <vector> #include <vector>
@@ -12,6 +11,7 @@
#include "spore/types/NodeInfo.h" #include "spore/types/NodeInfo.h"
#include "spore/core/TaskManager.h" #include "spore/core/TaskManager.h"
#include "spore/types/ApiTypes.h" #include "spore/types/ApiTypes.h"
#include "spore/util/Logging.h"
class Service; // Forward declaration class Service; // Forward declaration
@@ -20,15 +20,17 @@ public:
ApiServer(NodeContext& ctx, TaskManager& taskMgr, uint16_t port = 80); ApiServer(NodeContext& ctx, TaskManager& taskMgr, uint16_t port = 80);
void begin(); void begin();
void addService(Service& service); void addService(Service& service);
void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler);
void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler, void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler); const String& serviceName = "unknown");
void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
const std::vector<ParamSpec>& params);
void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler, void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler, std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler,
const std::vector<ParamSpec>& params); const String& serviceName = "unknown");
void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
const std::vector<ParamSpec>& params, const String& serviceName = "unknown");
void addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler,
const std::vector<ParamSpec>& params, const String& serviceName = "unknown");
// Static file serving // Static file serving
void serveStatic(const String& uri, fs::FS& fs, const String& path, const String& cache_header = ""); void serveStatic(const String& uri, fs::FS& fs, const String& path, const String& cache_header = "");
@@ -40,18 +42,13 @@ public:
private: private:
AsyncWebServer server; AsyncWebServer server;
AsyncWebSocket ws{ "/ws" };
NodeContext& ctx; NodeContext& ctx;
TaskManager& taskManager; TaskManager& taskManager;
std::vector<std::reference_wrapper<Service>> services; std::vector<std::reference_wrapper<Service>> services;
std::vector<EndpointInfo> endpoints; // Single source of truth for endpoints std::vector<EndpointInfo> endpoints; // Single source of truth for endpoints
std::vector<AsyncWebSocketClient*> wsClients;
// Internal helpers // Internal helpers
void registerEndpoint(const String& uri, int method, void registerEndpoint(const String& uri, int method,
const std::vector<ParamSpec>& params, const std::vector<ParamSpec>& params,
const String& serviceName); const String& serviceName);
// WebSocket helpers
void setupWebSocket();
}; };

View File

@@ -7,15 +7,13 @@
#include <ArduinoJson.h> #include <ArduinoJson.h>
#include <ESP8266HTTPClient.h> #include <ESP8266HTTPClient.h>
#include <map> #include <map>
#include <vector>
#include <functional>
class ClusterManager { class ClusterManager {
public: public:
ClusterManager(NodeContext& ctx, TaskManager& taskMgr); ClusterManager(NodeContext& ctx, TaskManager& taskMgr);
void registerTasks(); void registerTasks();
void sendDiscovery(); void sendDiscovery();
void listen(); void listenForDiscovery();
void addOrUpdateNode(const String& nodeHost, IPAddress nodeIP); void addOrUpdateNode(const String& nodeHost, IPAddress nodeIP);
void updateAllNodeStatuses(); void updateAllNodeStatuses();
void removeDeadNodes(); void removeDeadNodes();
@@ -28,23 +26,4 @@ public:
private: private:
NodeContext& ctx; NodeContext& ctx;
TaskManager& taskManager; TaskManager& taskManager;
struct MessageHandler {
bool (*predicate)(const char*);
std::function<void(const char*)> handle;
const char* name;
};
void initMessageHandlers();
void handleIncomingMessage(const char* incoming);
static bool isDiscoveryMsg(const char* msg);
static bool isHeartbeatMsg(const char* msg);
static bool isResponseMsg(const char* msg);
static bool isNodeInfoMsg(const char* msg);
static bool isClusterEventMsg(const char* msg);
void onDiscovery(const char* msg);
void onHeartbeat(const char* msg);
void onResponse(const char* msg);
void onNodeInfo(const char* msg);
void onClusterEvent(const char* msg);
unsigned long lastHeartbeatSentAt = 0;
std::vector<MessageHandler> messageHandlers;
}; };

View File

@@ -23,10 +23,7 @@ public:
using EventCallback = std::function<void(void*)>; using EventCallback = std::function<void(void*)>;
std::map<std::string, std::vector<EventCallback>> eventRegistry; std::map<std::string, std::vector<EventCallback>> eventRegistry;
using AnyEventCallback = std::function<void(const std::string&, void*)>;
std::vector<AnyEventCallback> anyEventSubscribers;
void on(const std::string& event, EventCallback cb); void on(const std::string& event, EventCallback cb);
void fire(const std::string& event, void* data); void fire(const std::string& event, void* data);
void onAny(AnyEventCallback cb);
}; };

View File

@@ -7,12 +7,8 @@
namespace ClusterProtocol { namespace ClusterProtocol {
constexpr const char* DISCOVERY_MSG = "CLUSTER_DISCOVERY"; constexpr const char* DISCOVERY_MSG = "CLUSTER_DISCOVERY";
constexpr const char* RESPONSE_MSG = "CLUSTER_RESPONSE"; constexpr const char* RESPONSE_MSG = "CLUSTER_RESPONSE";
constexpr const char* HEARTBEAT_MSG = "CLUSTER_HEARTBEAT";
constexpr const char* NODE_INFO_MSG = "CLUSTER_NODE_INFO";
constexpr const char* CLUSTER_EVENT_MSG = "CLUSTER_EVENT";
constexpr uint16_t UDP_PORT = 4210; constexpr uint16_t UDP_PORT = 4210;
// Increased buffer to accommodate node info JSON over UDP constexpr size_t UDP_BUF_SIZE = 64;
constexpr size_t UDP_BUF_SIZE = 512;
constexpr const char* API_NODE_STATUS = "/api/node/status"; constexpr const char* API_NODE_STATUS = "/api/node/status";
} }

View File

@@ -0,0 +1,114 @@
#pragma once
#include "spore/util/JsonSerializable.h"
#include "spore/util/Logging.h"
#include <ArduinoJson.h>
namespace spore {
namespace types {
/**
* Base class for API responses that can be serialized to JSON
* Handles complete JsonDocument creation and serialization
*/
class ApiResponse {
protected:
mutable JsonDocument doc;
public:
virtual ~ApiResponse() = default;
/**
* Get the complete JSON string representation of this response
* @return JSON string ready to send as HTTP response
*/
virtual String toJsonString() const {
String json;
serializeJson(doc, json);
return json;
}
/**
* Get the JsonDocument for direct manipulation if needed
* @return Reference to the internal JsonDocument
*/
JsonDocument& getDocument() { return doc; }
const JsonDocument& getDocument() const { return doc; }
/**
* Clear the document and reset for reuse
*/
virtual void clear() {
doc.clear();
}
};
/**
* Base class for API responses that contain a collection of serializable items
*/
template<typename ItemType>
class CollectionResponse : public ApiResponse {
protected:
String collectionKey;
public:
explicit CollectionResponse(const String& key) : collectionKey(key) {}
/**
* Add a serializable item to the collection
* @param item The serializable item to add
*/
void addItem(const util::JsonSerializable& item) {
// Ensure the array exists and get a reference to it
if (!doc[collectionKey].is<JsonArray>()) {
doc[collectionKey] = JsonArray();
}
JsonArray arr = doc[collectionKey].as<JsonArray>();
JsonObject obj = arr.add<JsonObject>();
item.toJson(obj);
}
/**
* Add a serializable item to the collection (move version)
* @param item The serializable item to add
*/
void addItem(util::JsonSerializable&& item) {
// Ensure the array exists and get a reference to it
if (!doc[collectionKey].is<JsonArray>()) {
doc[collectionKey] = JsonArray();
}
JsonArray arr = doc[collectionKey].as<JsonArray>();
JsonObject obj = arr.add<JsonObject>();
item.toJson(obj);
}
/**
* Add multiple items from a container
* @param items Container of serializable items
*/
template<typename Container>
void addItems(const Container& items) {
// Ensure the array exists and get a reference to it
if (!doc[collectionKey].is<JsonArray>()) {
doc[collectionKey] = JsonArray();
}
JsonArray arr = doc[collectionKey].as<JsonArray>();
for (const auto& item : items) {
JsonObject obj = arr.add<JsonObject>();
item.toJson(obj);
}
}
/**
* Get the current number of items in the collection
* @return Number of items
*/
size_t getItemCount() const {
if (doc[collectionKey].is<JsonArray>()) {
return doc[collectionKey].as<JsonArray>().size();
}
return 0;
}
};
} // namespace types
} // namespace spore

View File

@@ -0,0 +1,47 @@
#pragma once
#include "ApiResponse.h"
#include "NodeInfoSerializable.h"
#include <map>
namespace spore {
namespace types {
/**
* Response class for cluster members endpoint
* Handles complete JSON document creation for cluster member data
*/
class ClusterMembersResponse : public CollectionResponse<NodeInfoSerializable> {
public:
ClusterMembersResponse() : CollectionResponse("members") {}
/**
* Add a single node to the response
* @param node The NodeInfo to add
*/
void addNode(const NodeInfo& node) {
addItem(NodeInfoSerializable(const_cast<NodeInfo&>(node)));
}
/**
* Add multiple nodes from a member list
* @param memberList Map of hostname to NodeInfo
*/
void addNodes(const std::map<String, NodeInfo>& memberList) {
for (const auto& pair : memberList) {
addNode(pair.second);
}
}
/**
* Add nodes from a pointer to member list
* @param memberList Pointer to map of hostname to NodeInfo
*/
void addNodes(const std::map<String, NodeInfo>* memberList) {
if (memberList) {
addNodes(*memberList);
}
}
};
} // namespace types
} // namespace spore

View File

@@ -15,7 +15,6 @@ public:
// Cluster Configuration // Cluster Configuration
unsigned long discovery_interval_ms; unsigned long discovery_interval_ms;
unsigned long heartbeat_interval_ms; unsigned long heartbeat_interval_ms;
unsigned long cluster_listen_interval_ms;
unsigned long status_update_interval_ms; unsigned long status_update_interval_ms;
unsigned long member_info_update_interval_ms; unsigned long member_info_update_interval_ms;
unsigned long print_interval_ms; unsigned long print_interval_ms;

View File

@@ -0,0 +1,76 @@
#pragma once
#include "ApiTypes.h"
#include "spore/util/JsonSerializable.h"
#include <ArduinoJson.h>
namespace spore {
namespace types {
/**
* Serializable wrapper for EndpointInfo that implements JsonSerializable interface
* Handles conversion between EndpointInfo struct and JSON representation
*/
class EndpointInfoSerializable : public util::JsonSerializable {
private:
const EndpointInfo& endpoint;
public:
explicit EndpointInfoSerializable(const EndpointInfo& ep) : endpoint(ep) {}
/**
* Serialize EndpointInfo to JsonObject
*/
void toJson(JsonObject& obj) const override {
obj["uri"] = endpoint.uri;
obj["method"] = endpoint.method;
// Add parameters if present
if (!endpoint.params.empty()) {
JsonArray paramsArr = obj["params"].to<JsonArray>();
for (const auto& param : endpoint.params) {
JsonObject paramObj = paramsArr.add<JsonObject>();
paramObj["name"] = param.name;
paramObj["location"] = param.location;
paramObj["required"] = param.required;
paramObj["type"] = param.type;
if (!param.values.empty()) {
JsonArray valuesArr = paramObj["values"].to<JsonArray>();
for (const auto& value : param.values) {
valuesArr.add(value);
}
}
if (param.defaultValue.length() > 0) {
paramObj["default"] = param.defaultValue;
}
}
}
}
/**
* Deserialize EndpointInfo from JsonObject
*/
void fromJson(const JsonObject& obj) override {
// Note: This would require modifying the EndpointInfo struct to be mutable
// For now, this is a placeholder as EndpointInfo is typically read-only
// in the context where this serialization is used
}
};
/**
* Convenience function to create a JsonArray from a collection of EndpointInfo objects
*/
template<typename Container>
JsonArray endpointInfoToJsonArray(JsonDocument& doc, const Container& endpoints) {
JsonArray arr = doc.to<JsonArray>();
for (const auto& endpoint : endpoints) {
EndpointInfoSerializable serializable(endpoint);
JsonObject obj = arr.add<JsonObject>();
serializable.toJson(obj);
}
return arr;
}
} // namespace types
} // namespace spore

View File

@@ -17,7 +17,7 @@ struct NodeInfo {
uint32_t cpuFreqMHz = 0; uint32_t cpuFreqMHz = 0;
uint32_t flashChipSize = 0; uint32_t flashChipSize = 0;
} resources; } resources;
unsigned long latency = 0; // ms from heartbeat broadcast to NODE_INFO receipt unsigned long latency = 0; // ms since lastSeen
std::vector<EndpointInfo> endpoints; // List of registered endpoints std::vector<EndpointInfo> endpoints; // List of registered endpoints
std::map<String, String> labels; // Arbitrary node labels (key -> value) std::map<String, String> labels; // Arbitrary node labels (key -> value)
}; };

View File

@@ -0,0 +1,141 @@
#pragma once
#include "NodeInfo.h"
#include "ApiTypes.h"
#include "spore/util/JsonSerializable.h"
#include <ArduinoJson.h>
namespace spore {
namespace types {
/**
* Serializable wrapper for NodeInfo that implements JsonSerializable interface
* Handles conversion between NodeInfo struct and JSON representation
*/
class NodeInfoSerializable : public util::JsonSerializable {
private:
NodeInfo& nodeInfo;
public:
explicit NodeInfoSerializable(NodeInfo& node) : nodeInfo(node) {}
/**
* Serialize NodeInfo to JsonObject
* Maps all NodeInfo fields to appropriate JSON structure
*/
void toJson(JsonObject& obj) const override {
obj["hostname"] = nodeInfo.hostname;
obj["ip"] = nodeInfo.ip.toString();
obj["lastSeen"] = nodeInfo.lastSeen;
obj["latency"] = nodeInfo.latency;
obj["status"] = statusToStr(nodeInfo.status);
// Serialize resources
JsonObject resources = obj["resources"].to<JsonObject>();
resources["freeHeap"] = nodeInfo.resources.freeHeap;
resources["chipId"] = nodeInfo.resources.chipId;
resources["sdkVersion"] = nodeInfo.resources.sdkVersion;
resources["cpuFreqMHz"] = nodeInfo.resources.cpuFreqMHz;
resources["flashChipSize"] = nodeInfo.resources.flashChipSize;
// Serialize labels if present
if (!nodeInfo.labels.empty()) {
JsonObject labels = obj["labels"].to<JsonObject>();
for (const auto& kv : nodeInfo.labels) {
labels[kv.first.c_str()] = kv.second;
}
}
// Serialize endpoints if present
if (!nodeInfo.endpoints.empty()) {
JsonArray endpoints = obj["api"].to<JsonArray>();
for (const auto& endpoint : nodeInfo.endpoints) {
JsonObject endpointObj = endpoints.add<JsonObject>();
endpointObj["uri"] = endpoint.uri;
endpointObj["method"] = endpoint.method;
}
}
}
/**
* Deserialize NodeInfo from JsonObject
* Populates NodeInfo fields from JSON structure
*/
void fromJson(const JsonObject& obj) override {
nodeInfo.hostname = obj["hostname"].as<String>();
// Parse IP address
const char* ipStr = obj["ip"];
if (ipStr) {
nodeInfo.ip.fromString(ipStr);
}
nodeInfo.lastSeen = obj["lastSeen"].as<unsigned long>();
nodeInfo.latency = obj["latency"].as<unsigned long>();
// Parse status
const char* statusStr = obj["status"];
if (statusStr) {
if (strcmp(statusStr, "ACTIVE") == 0) {
nodeInfo.status = NodeInfo::ACTIVE;
} else if (strcmp(statusStr, "INACTIVE") == 0) {
nodeInfo.status = NodeInfo::INACTIVE;
} else if (strcmp(statusStr, "DEAD") == 0) {
nodeInfo.status = NodeInfo::DEAD;
}
}
// Parse resources
if (obj["resources"].is<JsonObject>()) {
JsonObject resources = obj["resources"].as<JsonObject>();
nodeInfo.resources.freeHeap = resources["freeHeap"].as<uint32_t>();
nodeInfo.resources.chipId = resources["chipId"].as<uint32_t>();
nodeInfo.resources.sdkVersion = resources["sdkVersion"].as<String>();
nodeInfo.resources.cpuFreqMHz = resources["cpuFreqMHz"].as<uint32_t>();
nodeInfo.resources.flashChipSize = resources["flashChipSize"].as<uint32_t>();
}
// Parse labels
nodeInfo.labels.clear();
if (obj["labels"].is<JsonObject>()) {
JsonObject labels = obj["labels"].as<JsonObject>();
for (JsonPair kvp : labels) {
nodeInfo.labels[kvp.key().c_str()] = kvp.value().as<String>();
}
}
// Parse endpoints
nodeInfo.endpoints.clear();
if (obj["api"].is<JsonArray>()) {
JsonArray endpoints = obj["api"].as<JsonArray>();
for (JsonObject endpointObj : endpoints) {
EndpointInfo endpoint;
endpoint.uri = endpointObj["uri"].as<String>();
endpoint.method = endpointObj["method"].as<int>();
endpoint.isLocal = false;
endpoint.serviceName = "remote";
nodeInfo.endpoints.push_back(std::move(endpoint));
}
}
}
};
/**
* Convenience function to create a JsonArray from a collection of NodeInfo objects
* @param doc The JsonDocument to create the array in
* @param nodes Collection of NodeInfo objects
* @return A JsonArray containing all serialized NodeInfo objects
*/
template<typename Container>
JsonArray nodeInfoToJsonArray(JsonDocument& doc, const Container& nodes) {
JsonArray arr = doc.to<JsonArray>();
for (const auto& pair : nodes) {
const NodeInfo& node = pair.second;
NodeInfoSerializable serializable(const_cast<NodeInfo&>(node));
JsonObject obj = arr.add<JsonObject>();
serializable.toJson(obj);
}
return arr;
}
} // namespace types
} // namespace spore

View File

@@ -0,0 +1,102 @@
#pragma once
#include "ApiResponse.h"
#include "EndpointInfoSerializable.h"
#include "NodeInfo.h"
#include "spore/util/Logging.h"
#include <vector>
namespace spore {
namespace types {
/**
* Response class for node status endpoint
* Handles complete JSON document creation for node status data
*/
class NodeStatusResponse : public ApiResponse {
public:
/**
* Set basic system information
*/
void setSystemInfo() {
doc["freeHeap"] = ESP.getFreeHeap();
doc["chipId"] = ESP.getChipId();
doc["sdkVersion"] = ESP.getSdkVersion();
doc["cpuFreqMHz"] = ESP.getCpuFreqMHz();
doc["flashChipSize"] = ESP.getFlashChipSize();
}
/**
* Add labels to the response
* @param labels Map of label key-value pairs
*/
void addLabels(const std::map<String, String>& labels) {
if (!labels.empty()) {
JsonObject labelsObj = doc["labels"].to<JsonObject>();
for (const auto& kv : labels) {
labelsObj[kv.first.c_str()] = kv.second;
}
}
}
/**
* Build complete response with system info and labels
* @param labels Optional labels to include
*/
void buildCompleteResponse(const std::map<String, String>& labels = {}) {
setSystemInfo();
addLabels(labels);
}
};
/**
* Response class for node endpoints endpoint
* Handles complete JSON document creation for endpoint data
*/
class NodeEndpointsResponse : public CollectionResponse<EndpointInfoSerializable> {
public:
NodeEndpointsResponse() : CollectionResponse("endpoints") {}
/**
* Add a single endpoint to the response
* @param endpoint The EndpointInfo to add
*/
void addEndpoint(const EndpointInfo& endpoint) {
addItem(EndpointInfoSerializable(endpoint));
}
/**
* Add multiple endpoints from a container
* @param endpoints Container of EndpointInfo objects
*/
void addEndpoints(const std::vector<EndpointInfo>& endpoints) {
for (const auto& endpoint : endpoints) {
addEndpoint(endpoint);
}
}
};
/**
* Response class for simple status operations (update, restart)
* Handles simple JSON responses for node operations
*/
class NodeOperationResponse : public ApiResponse {
public:
/**
* Set success response
* @param status Status message
*/
void setSuccess(const String& status) {
doc["status"] = status;
}
/**
* Set error response
* @param status Error status message
*/
void setError(const String& status) {
doc["status"] = status;
}
};
} // namespace types
} // namespace spore

View File

@@ -0,0 +1,99 @@
#pragma once
#include "spore/util/JsonSerializable.h"
#include <ArduinoJson.h>
#include <Arduino.h>
namespace spore {
namespace types {
/**
* Serializable wrapper for task information that implements JsonSerializable interface
* Handles conversion between task data and JSON representation
*/
class TaskInfoSerializable : public util::JsonSerializable {
private:
String taskName;
const JsonObject& taskData;
public:
TaskInfoSerializable(const String& name, const JsonObject& data)
: taskName(name), taskData(data) {}
TaskInfoSerializable(const std::string& name, const JsonObject& data)
: taskName(name.c_str()), taskData(data) {}
/**
* Serialize task info to JsonObject
*/
void toJson(JsonObject& obj) const override {
obj["name"] = taskName;
obj["interval"] = taskData["interval"];
obj["enabled"] = taskData["enabled"];
obj["running"] = taskData["running"];
obj["autoStart"] = taskData["autoStart"];
}
/**
* Deserialize task info from JsonObject
* Note: This is read-only for task status, so fromJson is not implemented
*/
void fromJson(const JsonObject& obj) override {
// Task info is typically read-only in this context
// Implementation would go here if needed
}
};
/**
* Serializable wrapper for system information
*/
class SystemInfoSerializable : public util::JsonSerializable {
public:
/**
* Serialize system info to JsonObject
*/
void toJson(JsonObject& obj) const override {
obj["freeHeap"] = ESP.getFreeHeap();
obj["uptime"] = millis();
}
/**
* Deserialize system info from JsonObject
* Note: System info is typically read-only, so fromJson is not implemented
*/
void fromJson(const JsonObject& obj) override {
// System info is typically read-only
// Implementation would go here if needed
}
};
/**
* Serializable wrapper for task summary information
*/
class TaskSummarySerializable : public util::JsonSerializable {
private:
size_t totalTasks;
size_t activeTasks;
public:
TaskSummarySerializable(size_t total, size_t active)
: totalTasks(total), activeTasks(active) {}
/**
* Serialize task summary to JsonObject
*/
void toJson(JsonObject& obj) const override {
obj["totalTasks"] = totalTasks;
obj["activeTasks"] = activeTasks;
}
/**
* Deserialize task summary from JsonObject
*/
void fromJson(const JsonObject& obj) override {
totalTasks = obj["totalTasks"].as<size_t>();
activeTasks = obj["activeTasks"].as<size_t>();
}
};
} // namespace types
} // namespace spore

View File

@@ -0,0 +1,194 @@
#pragma once
#include "ApiResponse.h"
#include "TaskInfoSerializable.h"
#include <map>
namespace spore {
namespace types {
/**
* Response class for task status endpoint
* Handles complete JSON document creation for task status data
*/
class TaskStatusResponse : public ApiResponse {
public:
/**
* Set the task summary information
* @param totalTasks Total number of tasks
* @param activeTasks Number of active tasks
*/
void setSummary(size_t totalTasks, size_t activeTasks) {
TaskSummarySerializable summary(totalTasks, activeTasks);
JsonObject summaryObj = doc["summary"].to<JsonObject>();
summary.toJson(summaryObj);
}
/**
* Add a single task to the response
* @param taskName Name of the task
* @param taskData Task data as JsonObject
*/
void addTask(const String& taskName, const JsonObject& taskData) {
TaskInfoSerializable serializable(taskName, taskData);
if (!doc["tasks"].is<JsonArray>()) {
doc["tasks"] = JsonArray();
}
JsonArray tasksArr = doc["tasks"].as<JsonArray>();
JsonObject taskObj = tasksArr.add<JsonObject>();
serializable.toJson(taskObj);
}
/**
* Add a single task to the response (std::string version)
* @param taskName Name of the task
* @param taskData Task data as JsonObject
*/
void addTask(const std::string& taskName, const JsonObject& taskData) {
TaskInfoSerializable serializable(taskName, taskData);
if (!doc["tasks"].is<JsonArray>()) {
doc["tasks"] = JsonArray();
}
JsonArray tasksArr = doc["tasks"].as<JsonArray>();
JsonObject taskObj = tasksArr.add<JsonObject>();
serializable.toJson(taskObj);
}
/**
* Add multiple tasks from a task statuses map
* @param taskStatuses Map of task name to task data
*/
void addTasks(const std::map<String, JsonObject>& taskStatuses) {
for (const auto& pair : taskStatuses) {
addTask(pair.first, pair.second);
}
}
/**
* Set the system information
*/
void setSystemInfo() {
SystemInfoSerializable systemInfo;
JsonObject systemObj = doc["system"].to<JsonObject>();
systemInfo.toJson(systemObj);
}
/**
* Build complete response with all components
* @param taskStatuses Map of task name to task data
*/
void buildCompleteResponse(const std::map<String, JsonObject>& taskStatuses) {
// Set summary
size_t totalTasks = taskStatuses.size();
size_t activeTasks = 0;
for (const auto& pair : taskStatuses) {
if (pair.second["enabled"]) {
activeTasks++;
}
}
setSummary(totalTasks, activeTasks);
// Add all tasks
addTasks(taskStatuses);
// Set system info
setSystemInfo();
}
/**
* Build complete response with all components from vector
* @param taskStatuses Vector of pairs of task name to task data
*/
void buildCompleteResponse(const std::vector<std::pair<std::string, JsonObject>>& taskStatuses) {
// Clear the document first since getAllTaskStatuses creates a root array
doc.clear();
// Set summary
size_t totalTasks = taskStatuses.size();
size_t activeTasks = 0;
for (const auto& pair : taskStatuses) {
if (pair.second["enabled"]) {
activeTasks++;
}
}
setSummary(totalTasks, activeTasks);
// Add all tasks - extract data before clearing to avoid invalid references
JsonArray tasksArr = doc["tasks"].to<JsonArray>();
for (const auto& pair : taskStatuses) {
// Extract data from JsonObject before it becomes invalid
String taskName = pair.first.c_str();
unsigned long interval = pair.second["interval"];
bool enabled = pair.second["enabled"];
bool running = pair.second["running"];
bool autoStart = pair.second["autoStart"];
// Create new JsonObject in our document
JsonObject taskObj = tasksArr.add<JsonObject>();
taskObj["name"] = taskName;
taskObj["interval"] = interval;
taskObj["enabled"] = enabled;
taskObj["running"] = running;
taskObj["autoStart"] = autoStart;
}
// Set system info
setSystemInfo();
}
};
/**
* Response class for task control operations
* Handles JSON responses for task enable/disable/start/stop operations
*/
class TaskControlResponse : public ApiResponse {
public:
/**
* Set the response data for a task control operation
* @param success Whether the operation was successful
* @param message Response message
* @param taskName Name of the task
* @param action Action performed
*/
void setResponse(bool success, const String& message, const String& taskName, const String& action) {
doc["success"] = success;
doc["message"] = message;
doc["task"] = taskName;
doc["action"] = action;
}
/**
* Add detailed task information to the response
* @param taskName Name of the task
* @param enabled Whether task is enabled
* @param running Whether task is running
* @param interval Task interval
*/
void addTaskDetails(const String& taskName, bool enabled, bool running, unsigned long interval) {
JsonObject taskDetails = doc["taskDetails"].to<JsonObject>();
taskDetails["name"] = taskName;
taskDetails["enabled"] = enabled;
taskDetails["running"] = running;
taskDetails["interval"] = interval;
// Add system info
SystemInfoSerializable systemInfo;
JsonObject systemObj = taskDetails["system"].to<JsonObject>();
systemInfo.toJson(systemObj);
}
/**
* Set error response
* @param message Error message
* @param example Optional example for correct usage
*/
void setError(const String& message, const String& example = "") {
doc["success"] = false;
doc["message"] = message;
if (example.length() > 0) {
doc["example"] = example;
}
}
};
} // namespace types
} // namespace spore

View File

@@ -0,0 +1,56 @@
#pragma once
#include <ArduinoJson.h>
namespace spore {
namespace util {
/**
* Abstract base class for objects that can be serialized to/from JSON
* Provides a clean interface for converting objects to JsonObject and back
*/
class JsonSerializable {
public:
virtual ~JsonSerializable() = default;
/**
* Serialize this object to a JsonObject
* @param obj The JsonObject to populate with this object's data
*/
virtual void toJson(JsonObject& obj) const = 0;
/**
* Deserialize this object from a JsonObject
* @param obj The JsonObject containing the data to populate this object
*/
virtual void fromJson(const JsonObject& obj) = 0;
/**
* Convenience method to create a JsonObject from this object
* @param doc The JsonDocument to create the object in
* @return A JsonObject containing this object's serialized data
*/
JsonObject toJsonObject(JsonDocument& doc) const {
JsonObject obj = doc.to<JsonObject>();
toJson(obj);
return obj;
}
/**
* Convenience method to create a JsonArray from a collection of serializable objects
* @param doc The JsonDocument to create the array in
* @param objects Collection of objects implementing JsonSerializable
* @return A JsonArray containing all serialized objects
*/
template<typename Container>
static JsonArray toJsonArray(JsonDocument& doc, const Container& objects) {
JsonArray arr = doc.to<JsonArray>();
for (const auto& obj : objects) {
JsonObject item = arr.add<JsonObject>();
obj.toJson(item);
}
return arr;
}
};
} // namespace util
} // namespace spore

View File

@@ -21,57 +21,31 @@ void ApiServer::registerEndpoint(const String& uri, int method,
const String& serviceName) { const String& serviceName) {
// Add to local endpoints // Add to local endpoints
endpoints.push_back(EndpointInfo{uri, method, params, serviceName, true}); endpoints.push_back(EndpointInfo{uri, method, params, serviceName, true});
// Update cluster if needed
if (ctx.memberList && !ctx.memberList->empty()) {
auto it = ctx.memberList->find(ctx.hostname);
if (it != ctx.memberList->end()) {
it->second.endpoints.push_back(EndpointInfo{uri, method, params, serviceName, true});
}
}
} }
void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler) { void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
// Get current service name if available const String& serviceName) {
String serviceName = "unknown";
if (!services.empty()) {
serviceName = services.back().get().getName();
}
registerEndpoint(uri, method, {}, serviceName); registerEndpoint(uri, method, {}, serviceName);
server.on(uri.c_str(), method, requestHandler); server.on(uri.c_str(), method, requestHandler);
} }
void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler, void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler) { std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler,
// Get current service name if available const String& serviceName) {
String serviceName = "unknown";
if (!services.empty()) {
serviceName = services.back().get().getName();
}
registerEndpoint(uri, method, {}, serviceName); registerEndpoint(uri, method, {}, serviceName);
server.on(uri.c_str(), method, requestHandler, uploadHandler); server.on(uri.c_str(), method, requestHandler, uploadHandler);
} }
// Overloads that also record minimal capability specs // Overloads that also record minimal capability specs
void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler, void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
const std::vector<ParamSpec>& params) { const std::vector<ParamSpec>& params, const String& serviceName) {
// Get current service name if available
String serviceName = "unknown";
if (!services.empty()) {
serviceName = services.back().get().getName();
}
registerEndpoint(uri, method, params, serviceName); registerEndpoint(uri, method, params, serviceName);
server.on(uri.c_str(), method, requestHandler); server.on(uri.c_str(), method, requestHandler);
} }
void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler, void ApiServer::addEndpoint(const String& uri, int method, std::function<void(AsyncWebServerRequest*)> requestHandler,
std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler, std::function<void(AsyncWebServerRequest*, const String&, size_t, uint8_t*, size_t, bool)> uploadHandler,
const std::vector<ParamSpec>& params) { const std::vector<ParamSpec>& params, const String& serviceName) {
// Get current service name if available
String serviceName = "unknown";
if (!services.empty()) {
serviceName = services.back().get().getName();
}
registerEndpoint(uri, method, params, serviceName); registerEndpoint(uri, method, params, serviceName);
server.on(uri.c_str(), method, requestHandler, uploadHandler); server.on(uri.c_str(), method, requestHandler, uploadHandler);
} }
@@ -87,10 +61,6 @@ void ApiServer::serveStatic(const String& uri, fs::FS& fs, const String& path, c
} }
void ApiServer::begin() { void ApiServer::begin() {
// Setup streaming API (WebSocket)
setupWebSocket();
server.addHandler(&ws);
// Register all service endpoints // Register all service endpoints
for (auto& service : services) { for (auto& service : services) {
service.get().registerEndpoints(*this); service.get().registerEndpoints(*this);
@@ -99,90 +69,3 @@ void ApiServer::begin() {
server.begin(); server.begin();
} }
void ApiServer::setupWebSocket() {
ws.onEvent([this](AsyncWebSocket* server, AsyncWebSocketClient* client, AwsEventType type, void* arg, uint8_t* data, size_t len) {
if (type == WS_EVT_DATA) {
AwsFrameInfo* info = (AwsFrameInfo*)arg;
if (info->final && info->index == 0 && info->len == len && info->opcode == WS_TEXT) {
// Parse directly from the raw buffer with explicit length
JsonDocument doc;
DeserializationError err = deserializeJson(doc, (const char*)data, len);
if (!err) {
LOG_DEBUG("API", "Received event: " + String(doc["event"].as<String>()));
String eventName = doc["event"].as<String>();
String payloadStr;
if (doc["payload"].is<const char*>()) {
payloadStr = doc["payload"].as<const char*>();
} else if (!doc["payload"].isNull()) {
// If payload is an object/array, serialize it
String tmp; serializeJson(doc["payload"], tmp); payloadStr = tmp;
}
// Allow empty payload; services may treat it as defaults
if (eventName.length() > 0) {
// Inject origin tag into payload JSON if possible
String enriched = payloadStr;
if (payloadStr.length() > 0) {
JsonDocument pd;
if (!deserializeJson(pd, payloadStr)) {
pd["_origin"] = String("ws:") + String(client->id());
String tmp; serializeJson(pd, tmp); enriched = tmp;
} else {
// If payload is plain string, leave as-is (no origin)
}
}
std::string ev = eventName.c_str();
ctx.fire(ev, &enriched);
// Acknowledge
client->text("{\"ok\":true}");
} else {
client->text("{\"error\":\"Missing 'event'\"}");
}
} else {
client->text("{\"error\":\"Invalid JSON\"}");
}
}
} else if (type == WS_EVT_CONNECT) {
client->text("{\"hello\":\"ws connected\"}");
wsClients.push_back(client);
} else if (type == WS_EVT_DISCONNECT) {
wsClients.erase(std::remove(wsClients.begin(), wsClients.end(), client), wsClients.end());
}
});
// Subscribe to all local events and forward to websocket clients
ctx.onAny([this](const std::string& event, void* dataPtr) {
String* payloadStrPtr = static_cast<String*>(dataPtr);
String payloadStr = payloadStrPtr ? *payloadStrPtr : String("");
// Extract and strip origin if present
String origin;
String cleanedPayload = payloadStr;
if (payloadStr.length() > 0) {
JsonDocument pd;
if (!deserializeJson(pd, payloadStr)) {
if (pd["_origin"].is<const char*>()) {
origin = pd["_origin"].as<const char*>();
pd.remove("_origin");
String tmp; serializeJson(pd, tmp); cleanedPayload = tmp;
}
}
}
JsonDocument outDoc;
outDoc["event"] = event.c_str();
outDoc["payload"] = cleanedPayload;
String out; serializeJson(outDoc, out);
if (origin.startsWith("ws:")) {
uint32_t originId = (uint32_t)origin.substring(3).toInt();
for (auto* c : wsClients) {
if (c && c->id() != originId) {
c->text(out);
}
}
} else {
ws.textAll(out);
}
});
}

View File

@@ -8,35 +8,17 @@ 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_DEBUG("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();
} }
void ClusterManager::registerTasks() { void ClusterManager::registerTasks() {
taskManager.registerTask("cluster_discovery", ctx.config.discovery_interval_ms, [this]() { sendDiscovery(); }); taskManager.registerTask("discovery_send", ctx.config.discovery_interval_ms, [this]() { sendDiscovery(); });
taskManager.registerTask("cluster_listen", ctx.config.cluster_listen_interval_ms, [this]() { listen(); }); taskManager.registerTask("discovery_listen", ctx.config.discovery_interval_ms / 10, [this]() { listenForDiscovery(); });
taskManager.registerTask("status_update", ctx.config.status_update_interval_ms, [this]() { updateAllNodeStatuses(); removeDeadNodes(); }); taskManager.registerTask("status_update", ctx.config.status_update_interval_ms, [this]() { updateAllNodeStatuses(); removeDeadNodes(); });
taskManager.registerTask("print_members", ctx.config.print_interval_ms, [this]() { printMemberList(); }); taskManager.registerTask("print_members", ctx.config.print_interval_ms, [this]() { printMemberList(); });
taskManager.registerTask("heartbeat", ctx.config.heartbeat_interval_ms, [this]() { heartbeatTaskCallback(); }); taskManager.registerTask("heartbeat", ctx.config.heartbeat_interval_ms, [this]() { heartbeatTaskCallback(); });
taskManager.registerTask("cluster_update_members_info", ctx.config.member_info_update_interval_ms, [this]() { updateAllMembersInfoTaskCallback(); }); taskManager.registerTask("update_members_info", ctx.config.member_info_update_interval_ms, [this]() { updateAllMembersInfoTaskCallback(); });
LOG_INFO("ClusterManager", "Registered all cluster tasks"); LOG_INFO("ClusterManager", "Registered all cluster tasks");
} }
@@ -47,210 +29,28 @@ void ClusterManager::sendDiscovery() {
ctx.udp->endPacket(); ctx.udp->endPacket();
} }
void ClusterManager::listen() { void ClusterManager::listenForDiscovery() {
int packetSize = ctx.udp->parsePacket(); int packetSize = ctx.udp->parsePacket();
if (!packetSize) { if (packetSize) {
return; char incoming[ClusterProtocol::UDP_BUF_SIZE];
} int len = ctx.udp->read(incoming, ClusterProtocol::UDP_BUF_SIZE);
if (len > 0) {
char incoming[ClusterProtocol::UDP_BUF_SIZE]; incoming[len] = 0;
int len = ctx.udp->read(incoming, ClusterProtocol::UDP_BUF_SIZE); }
if (len <= 0) { //LOG_DEBUG(ctx, "UDP", "Packet received: " + String(incoming));
return; if (strcmp(incoming, ClusterProtocol::DISCOVERY_MSG) == 0) {
} //LOG_DEBUG(ctx, "UDP", "Discovery request from: " + ctx.udp->remoteIP().toString());
if (len >= (int)ClusterProtocol::UDP_BUF_SIZE) { ctx.udp->beginPacket(ctx.udp->remoteIP(), ctx.config.udp_port);
incoming[ClusterProtocol::UDP_BUF_SIZE - 1] = 0; String response = String(ClusterProtocol::RESPONSE_MSG) + ":" + ctx.hostname;
} else { ctx.udp->write(response.c_str());
incoming[len] = 0; ctx.udp->endPacket();
} //LOG_DEBUG(ctx, "UDP", "Sent response with hostname: " + ctx.hostname);
handleIncomingMessage(incoming); } else if (strncmp(incoming, ClusterProtocol::RESPONSE_MSG, strlen(ClusterProtocol::RESPONSE_MSG)) == 0) {
} char* hostPtr = incoming + strlen(ClusterProtocol::RESPONSE_MSG) + 1;
String nodeHost = String(hostPtr);
void ClusterManager::initMessageHandlers() { addOrUpdateNode(nodeHost, ctx.udp->remoteIP());
messageHandlers.clear();
messageHandlers.push_back({ &ClusterManager::isDiscoveryMsg, [this](const char* msg){ this->onDiscovery(msg); }, "DISCOVERY" });
messageHandlers.push_back({ &ClusterManager::isHeartbeatMsg, [this](const char* msg){ this->onHeartbeat(msg); }, "HEARTBEAT" });
messageHandlers.push_back({ &ClusterManager::isResponseMsg, [this](const char* msg){ this->onResponse(msg); }, "RESPONSE" });
messageHandlers.push_back({ &ClusterManager::isNodeInfoMsg, [this](const char* msg){ this->onNodeInfo(msg); }, "NODE_INFO" });
messageHandlers.push_back({ &ClusterManager::isClusterEventMsg, [this](const char* msg){ this->onClusterEvent(msg); }, "CLUSTER_EVENT" });
}
void ClusterManager::handleIncomingMessage(const char* incoming) {
for (const auto& h : messageHandlers) {
if (h.predicate(incoming)) {
h.handle(incoming);
return;
} }
} }
// Unknown message - log first token
const char* colon = strchr(incoming, ':');
String head;
if (colon) {
head = String(incoming).substring(0, colon - incoming);
} else {
head = String(incoming);
}
LOG_DEBUG("Cluster", String("Unknown cluster message: ") + head);
}
bool ClusterManager::isDiscoveryMsg(const char* msg) {
return strcmp(msg, ClusterProtocol::DISCOVERY_MSG) == 0;
}
bool ClusterManager::isHeartbeatMsg(const char* msg) {
return strncmp(msg, ClusterProtocol::HEARTBEAT_MSG, strlen(ClusterProtocol::HEARTBEAT_MSG)) == 0;
}
bool ClusterManager::isResponseMsg(const char* msg) {
return strncmp(msg, ClusterProtocol::RESPONSE_MSG, strlen(ClusterProtocol::RESPONSE_MSG)) == 0;
}
bool ClusterManager::isNodeInfoMsg(const char* msg) {
return strncmp(msg, ClusterProtocol::NODE_INFO_MSG, strlen(ClusterProtocol::NODE_INFO_MSG)) == 0;
}
bool ClusterManager::isClusterEventMsg(const char* msg) {
return strncmp(msg, ClusterProtocol::CLUSTER_EVENT_MSG, strlen(ClusterProtocol::CLUSTER_EVENT_MSG)) == 0;
}
void ClusterManager::onDiscovery(const char* /*msg*/) {
ctx.udp->beginPacket(ctx.udp->remoteIP(), ctx.config.udp_port);
String response = String(ClusterProtocol::RESPONSE_MSG) + ":" + ctx.hostname;
ctx.udp->write(response.c_str());
ctx.udp->endPacket();
}
void ClusterManager::onHeartbeat(const char* /*msg*/) {
JsonDocument doc;
doc["freeHeap"] = ESP.getFreeHeap();
doc["chipId"] = ESP.getChipId();
doc["sdkVersion"] = ESP.getSdkVersion();
doc["cpuFreqMHz"] = ESP.getCpuFreqMHz();
doc["flashChipSize"] = ESP.getFlashChipSize();
if (ctx.memberList) {
auto it = ctx.memberList->find(ctx.hostname);
if (it != ctx.memberList->end()) {
JsonObject labelsObj = doc["labels"].to<JsonObject>();
for (const auto& kv : it->second.labels) {
labelsObj[kv.first.c_str()] = kv.second;
}
} else if (!ctx.self.labels.empty()) {
JsonObject labelsObj = doc["labels"].to<JsonObject>();
for (const auto& kv : ctx.self.labels) {
labelsObj[kv.first.c_str()] = kv.second;
}
}
}
String json;
serializeJson(doc, json);
ctx.udp->beginPacket(ctx.udp->remoteIP(), ctx.config.udp_port);
String msg = String(ClusterProtocol::NODE_INFO_MSG) + ":" + ctx.hostname + ":" + json;
ctx.udp->write(msg.c_str());
ctx.udp->endPacket();
}
void ClusterManager::onResponse(const char* msg) {
char* hostPtr = const_cast<char*>(msg) + strlen(ClusterProtocol::RESPONSE_MSG) + 1;
String nodeHost = String(hostPtr);
addOrUpdateNode(nodeHost, ctx.udp->remoteIP());
}
void ClusterManager::onNodeInfo(const char* msg) {
char* p = const_cast<char*>(msg) + strlen(ClusterProtocol::NODE_INFO_MSG) + 1;
char* hostEnd = strchr(p, ':');
if (hostEnd) {
*hostEnd = '\0';
const char* hostCStr = p;
const char* jsonCStr = hostEnd + 1;
String nodeHost = String(hostCStr);
IPAddress senderIP = ctx.udp->remoteIP();
addOrUpdateNode(nodeHost, senderIP);
JsonDocument doc;
DeserializationError err = deserializeJson(doc, jsonCStr);
if (!err) {
auto& memberList = *ctx.memberList;
auto it = memberList.find(nodeHost);
if (it != memberList.end()) {
NodeInfo& node = it->second;
node.resources.freeHeap = doc["freeHeap"] | node.resources.freeHeap;
node.resources.chipId = doc["chipId"] | node.resources.chipId;
{
const char* sdk = doc["sdkVersion"] | node.resources.sdkVersion.c_str();
node.resources.sdkVersion = sdk ? String(sdk) : node.resources.sdkVersion;
}
node.resources.cpuFreqMHz = doc["cpuFreqMHz"] | node.resources.cpuFreqMHz;
node.resources.flashChipSize = doc["flashChipSize"] | node.resources.flashChipSize;
node.status = NodeInfo::ACTIVE;
unsigned long now = millis();
node.lastSeen = now;
if (lastHeartbeatSentAt != 0) {
node.latency = now - lastHeartbeatSentAt;
}
node.labels.clear();
if (doc["labels"].is<JsonObject>()) {
JsonObject labelsObj = doc["labels"].as<JsonObject>();
for (JsonPair kvp : labelsObj) {
const char* key = kvp.key().c_str();
const char* value = labelsObj[kvp.key()];
node.labels[key] = value;
}
}
}
} else {
LOG_WARN("Cluster", String("Failed to parse NODE_INFO JSON from ") + senderIP.toString());
}
}
}
void ClusterManager::onClusterEvent(const char* msg) {
// Message format: CLUSTER_EVENT:{"event":"...","data":"<json string>"}
const char* jsonStart = msg + strlen(ClusterProtocol::CLUSTER_EVENT_MSG) + 1; // skip prefix and ':'
if (*jsonStart == '\0') {
LOG_DEBUG("Cluster", "CLUSTER_EVENT received with empty payload");
return;
}
LOG_DEBUG("Cluster", String("CLUSTER_EVENT raw from ") + ctx.udp->remoteIP().toString() + " len=" + String(strlen(jsonStart)));
JsonDocument doc;
DeserializationError err = deserializeJson(doc, jsonStart);
if (err) {
LOG_ERROR("Cluster", String("Failed to parse CLUSTER_EVENT JSON from ") + ctx.udp->remoteIP().toString());
return;
}
// Robust extraction of event and data
String eventStr;
if (doc["event"].is<const char*>()) {
eventStr = doc["event"].as<const char*>();
} else if (doc["event"].is<String>()) {
eventStr = doc["event"].as<String>();
}
String data;
if (doc["data"].is<const char*>()) {
data = doc["data"].as<const char*>();
} else if (doc["data"].is<JsonVariantConst>()) {
// If data is a nested JSON object/array, serialize it back to string
String tmp;
serializeJson(doc["data"], tmp);
data = tmp;
}
if (eventStr.length() == 0 || data.length() == 0) {
String dbg;
serializeJson(doc, dbg);
LOG_WARN("Cluster", String("CLUSTER_EVENT missing 'event' or 'data' | payload=") + dbg);
return;
}
std::string eventKey(eventStr.c_str());
LOG_DEBUG("Cluster", String("Firing event '") + eventStr + "' with dataLen=" + String(data.length()));
ctx.fire(eventKey, &data);
} }
void ClusterManager::addOrUpdateNode(const String& nodeHost, IPAddress nodeIP) { void ClusterManager::addOrUpdateNode(const String& nodeHost, IPAddress nodeIP) {
@@ -271,7 +71,7 @@ void ClusterManager::addOrUpdateNode(const String& nodeHost, IPAddress nodeIP) {
newNode.hostname = nodeHost; newNode.hostname = nodeHost;
newNode.ip = nodeIP; newNode.ip = nodeIP;
newNode.lastSeen = millis(); newNode.lastSeen = millis();
updateNodeStatus(newNode, newNode.lastSeen, ctx.config.node_inactive_threshold_ms, ctx.config.node_dead_threshold_ms); updateNodeStatus(newNode, newNode.lastSeen, ctx.config.node_inactive_threshold_ms, ctx.config.node_dead_threshold_ms);
memberList[nodeHost] = newNode; memberList[nodeHost] = newNode;
LOG_INFO("Cluster", "Added node: " + nodeHost + " @ " + newNode.ip.toString() + " | Status: " + statusToStr(newNode.status) + " | last update: 0"); LOG_INFO("Cluster", "Added node: " + nodeHost + " @ " + newNode.ip.toString() + " | Status: " + statusToStr(newNode.status) + " | last update: 0");
//fetchNodeInfo(nodeIP); // Do not fetch here, handled by periodic task //fetchNodeInfo(nodeIP); // Do not fetch here, handled by periodic task
@@ -394,18 +194,31 @@ void ClusterManager::heartbeatTaskCallback() {
updateLocalNodeResources(); updateLocalNodeResources();
ctx.fire("node_discovered", &node); ctx.fire("node_discovered", &node);
} }
// Broadcast heartbeat so peers can respond with their node info
lastHeartbeatSentAt = millis();
ctx.udp->beginPacket("255.255.255.255", ctx.config.udp_port);
String hb = String(ClusterProtocol::HEARTBEAT_MSG) + ":" + ctx.hostname;
ctx.udp->write(hb.c_str());
ctx.udp->endPacket();
} }
void ClusterManager::updateAllMembersInfoTaskCallback() { void ClusterManager::updateAllMembersInfoTaskCallback() {
// HTTP-based member info fetching disabled; node info is provided via UDP responses to heartbeats auto& memberList = *ctx.memberList;
// No-op to reduce network and memory usage
// Limit concurrent HTTP requests to prevent memory pressure
const size_t maxConcurrentRequests = ctx.config.max_concurrent_http_requests;
size_t requestCount = 0;
for (auto& pair : memberList) {
const NodeInfo& node = pair.second;
if (node.ip != ctx.localIP) {
// Only process a limited number of requests per cycle
if (requestCount >= maxConcurrentRequests) {
LOG_DEBUG("Cluster", "Limiting concurrent HTTP requests to prevent memory pressure");
break;
}
fetchNodeInfo(node.ip);
requestCount++;
// Add small delay between requests to prevent overwhelming the system
delay(100);
}
}
} }
void ClusterManager::updateAllNodeStatuses() { void ClusterManager::updateAllNodeStatuses() {

View File

@@ -29,11 +29,4 @@ void NodeContext::fire(const std::string& event, void* data) {
for (auto& cb : eventRegistry[event]) { for (auto& cb : eventRegistry[event]) {
cb(data); cb(data);
} }
for (auto& acb : anyEventSubscribers) {
acb(event, data);
}
}
void NodeContext::onAny(AnyEventCallback cb) {
anyEventSubscribers.push_back(cb);
} }

View File

@@ -1,65 +1,19 @@
#include "spore/services/ClusterService.h" #include "spore/services/ClusterService.h"
#include "spore/core/ApiServer.h" #include "spore/core/ApiServer.h"
#include "spore/types/ClusterResponse.h"
using spore::types::ClusterMembersResponse;
ClusterService::ClusterService(NodeContext& ctx) : ctx(ctx) {} ClusterService::ClusterService(NodeContext& ctx) : ctx(ctx) {}
void ClusterService::registerEndpoints(ApiServer& api) { void ClusterService::registerEndpoints(ApiServer& api) {
api.addEndpoint("/api/cluster/members", HTTP_GET, api.addEndpoint("/api/cluster/members", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleMembersRequest(request); }, [this](AsyncWebServerRequest* request) { handleMembersRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "ClusterService");
// Generic cluster broadcast endpoint
api.addEndpoint("/api/cluster/event", HTTP_POST,
[this](AsyncWebServerRequest* request) {
if (!request->hasParam("event", true) || !request->hasParam("payload", true)) {
request->send(400, "application/json", "{\"error\":\"Missing 'event' or 'payload'\"}");
return;
}
String eventName = request->getParam("event", true)->value();
String payloadStr = request->getParam("payload", true)->value();
JsonDocument envelope;
envelope["event"] = eventName;
envelope["data"] = payloadStr; // pass payload as JSON string
String eventJson;
serializeJson(envelope, eventJson);
std::string ev = "cluster/broadcast";
ctx.fire(ev, &eventJson);
request->send(200, "application/json", "{\"ok\":true}");
},
std::vector<ParamSpec>{
ParamSpec{String("event"), true, String("body"), String("string"), {}},
ParamSpec{String("payload"), true, String("body"), String("string"), {}}
});
} }
void ClusterService::handleMembersRequest(AsyncWebServerRequest* request) { void ClusterService::handleMembersRequest(AsyncWebServerRequest* request) {
JsonDocument doc; ClusterMembersResponse response;
JsonArray arr = doc["members"].to<JsonArray>(); response.addNodes(ctx.memberList);
request->send(200, "application/json", response.toJsonString());
for (const auto& pair : *ctx.memberList) {
const NodeInfo& node = pair.second;
JsonObject obj = arr.add<JsonObject>();
obj["hostname"] = node.hostname;
obj["ip"] = node.ip.toString();
obj["lastSeen"] = node.lastSeen;
obj["latency"] = node.latency;
obj["status"] = statusToStr(node.status);
obj["resources"]["freeHeap"] = node.resources.freeHeap;
obj["resources"]["chipId"] = node.resources.chipId;
obj["resources"]["sdkVersion"] = node.resources.sdkVersion;
obj["resources"]["cpuFreqMHz"] = node.resources.cpuFreqMHz;
obj["resources"]["flashChipSize"] = node.resources.flashChipSize;
// Add labels if present
if (!node.labels.empty()) {
JsonObject labelsObj = obj["labels"].to<JsonObject>();
for (const auto& kv : node.labels) {
labelsObj[kv.first.c_str()] = kv.second;
}
}
}
String json;
serializeJson(doc, json);
request->send(200, "application/json", json);
} }

View File

@@ -12,7 +12,7 @@ MonitoringService::MonitoringService(CpuUsage& cpuUsage)
void MonitoringService::registerEndpoints(ApiServer& api) { void MonitoringService::registerEndpoints(ApiServer& api) {
api.addEndpoint("/api/monitoring/resources", HTTP_GET, api.addEndpoint("/api/monitoring/resources", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleResourcesRequest(request); }, [this](AsyncWebServerRequest* request) { handleResourcesRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "MonitoringService");
} }
MonitoringService::SystemResources MonitoringService::getSystemResources() const { MonitoringService::SystemResources MonitoringService::getSystemResources() const {

View File

@@ -8,16 +8,16 @@ void NetworkService::registerEndpoints(ApiServer& api) {
// WiFi scanning endpoints // WiFi scanning endpoints
api.addEndpoint("/api/network/wifi/scan", HTTP_POST, api.addEndpoint("/api/network/wifi/scan", HTTP_POST,
[this](AsyncWebServerRequest* request) { handleWifiScanRequest(request); }, [this](AsyncWebServerRequest* request) { handleWifiScanRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "NetworkService");
api.addEndpoint("/api/network/wifi/scan", HTTP_GET, api.addEndpoint("/api/network/wifi/scan", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleGetWifiNetworks(request); }, [this](AsyncWebServerRequest* request) { handleGetWifiNetworks(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "NetworkService");
// Network status and configuration endpoints // Network status and configuration endpoints
api.addEndpoint("/api/network/status", HTTP_GET, api.addEndpoint("/api/network/status", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleNetworkStatus(request); }, [this](AsyncWebServerRequest* request) { handleNetworkStatus(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "NetworkService");
api.addEndpoint("/api/network/wifi/config", HTTP_POST, api.addEndpoint("/api/network/wifi/config", HTTP_POST,
[this](AsyncWebServerRequest* request) { handleSetWifiConfig(request); }, [this](AsyncWebServerRequest* request) { handleSetWifiConfig(request); },
@@ -26,7 +26,7 @@ void NetworkService::registerEndpoints(ApiServer& api) {
ParamSpec{String("password"), true, String("body"), String("string"), {}, String("")}, ParamSpec{String("password"), true, String("body"), String("string"), {}, String("")},
ParamSpec{String("connect_timeout_ms"), false, String("body"), String("number"), {}, String("10000")}, ParamSpec{String("connect_timeout_ms"), false, String("body"), String("number"), {}, String("10000")},
ParamSpec{String("retry_delay_ms"), false, String("body"), String("number"), {}, String("500")} ParamSpec{String("retry_delay_ms"), false, String("body"), String("number"), {}, String("500")}
}); }, "NetworkService");
} }
void NetworkService::handleWifiScanRequest(AsyncWebServerRequest* request) { void NetworkService::handleWifiScanRequest(AsyncWebServerRequest* request) {

View File

@@ -1,6 +1,11 @@
#include "spore/services/NodeService.h" #include "spore/services/NodeService.h"
#include "spore/core/ApiServer.h" #include "spore/core/ApiServer.h"
#include "spore/util/Logging.h" #include "spore/util/Logging.h"
#include "spore/types/NodeResponse.h"
using spore::types::NodeStatusResponse;
using spore::types::NodeOperationResponse;
using spore::types::NodeEndpointsResponse;
NodeService::NodeService(NodeContext& ctx, ApiServer& apiServer) : ctx(ctx), apiServer(apiServer) {} NodeService::NodeService(NodeContext& ctx, ApiServer& apiServer) : ctx(ctx), apiServer(apiServer) {}
@@ -8,7 +13,7 @@ void NodeService::registerEndpoints(ApiServer& api) {
// Status endpoint // Status endpoint
api.addEndpoint("/api/node/status", HTTP_GET, api.addEndpoint("/api/node/status", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleStatusRequest(request); }, [this](AsyncWebServerRequest* request) { handleStatusRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "NodeService");
// Update endpoint with file upload // Update endpoint with file upload
api.addEndpoint("/api/node/update", HTTP_POST, api.addEndpoint("/api/node/update", HTTP_POST,
@@ -18,72 +23,45 @@ void NodeService::registerEndpoints(ApiServer& api) {
}, },
std::vector<ParamSpec>{ std::vector<ParamSpec>{
ParamSpec{String("firmware"), true, String("body"), String("file"), {}, String("")} ParamSpec{String("firmware"), true, String("body"), String("file"), {}, String("")}
}); }, "NodeService");
// Restart endpoint // Restart endpoint
api.addEndpoint("/api/node/restart", HTTP_POST, api.addEndpoint("/api/node/restart", HTTP_POST,
[this](AsyncWebServerRequest* request) { handleRestartRequest(request); }, [this](AsyncWebServerRequest* request) { handleRestartRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "NodeService");
// Endpoints endpoint // Endpoints endpoint
api.addEndpoint("/api/node/endpoints", HTTP_GET, api.addEndpoint("/api/node/endpoints", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleEndpointsRequest(request); }, [this](AsyncWebServerRequest* request) { handleEndpointsRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "NodeService");
// Generic local event endpoint
api.addEndpoint("/api/node/event", HTTP_POST,
[this](AsyncWebServerRequest* request) {
if (!request->hasParam("event", true) || !request->hasParam("payload", true)) {
request->send(400, "application/json", "{\"error\":\"Missing 'event' or 'payload'\"}");
return;
}
String eventName = request->getParam("event", true)->value();
String payloadStr = request->getParam("payload", true)->value();
std::string ev = eventName.c_str();
ctx.fire(ev, &payloadStr);
request->send(200, "application/json", "{\"ok\":true}");
},
std::vector<ParamSpec>{
ParamSpec{String("event"), true, String("body"), String("string"), {}},
ParamSpec{String("payload"), true, String("body"), String("string"), {}}
});
} }
void NodeService::handleStatusRequest(AsyncWebServerRequest* request) { void NodeService::handleStatusRequest(AsyncWebServerRequest* request) {
JsonDocument doc; NodeStatusResponse response;
doc["freeHeap"] = ESP.getFreeHeap();
doc["chipId"] = ESP.getChipId(); // Get labels from member list or self
doc["sdkVersion"] = ESP.getSdkVersion(); std::map<String, String> labels;
doc["cpuFreqMHz"] = ESP.getCpuFreqMHz();
doc["flashChipSize"] = ESP.getFlashChipSize();
// Include local node labels if present
if (ctx.memberList) { if (ctx.memberList) {
auto it = ctx.memberList->find(ctx.hostname); auto it = ctx.memberList->find(ctx.hostname);
if (it != ctx.memberList->end()) { if (it != ctx.memberList->end()) {
JsonObject labelsObj = doc["labels"].to<JsonObject>(); labels = it->second.labels;
for (const auto& kv : it->second.labels) {
labelsObj[kv.first.c_str()] = kv.second;
}
} else if (!ctx.self.labels.empty()) { } else if (!ctx.self.labels.empty()) {
JsonObject labelsObj = doc["labels"].to<JsonObject>(); labels = ctx.self.labels;
for (const auto& kv : ctx.self.labels) {
labelsObj[kv.first.c_str()] = kv.second;
}
} }
} }
String json; response.buildCompleteResponse(labels);
serializeJson(doc, json); request->send(200, "application/json", response.toJsonString());
request->send(200, "application/json", json);
} }
void NodeService::handleUpdateRequest(AsyncWebServerRequest* request) { void NodeService::handleUpdateRequest(AsyncWebServerRequest* request) {
bool success = !Update.hasError(); bool success = !Update.hasError();
AsyncWebServerResponse* response = request->beginResponse(200, "application/json", NodeOperationResponse response;
success ? "{\"status\": \"OK\"}" : "{\"status\": \"FAIL\"}"); response.setSuccess(success ? "OK" : "FAIL");
response->addHeader("Connection", "close");
request->send(response); AsyncWebServerResponse* httpResponse = request->beginResponse(200, "application/json", response.toJsonString());
httpResponse->addHeader("Connection", "close");
request->send(httpResponse);
request->onDisconnect([this]() { request->onDisconnect([this]() {
LOG_INFO("API", "Restart device"); LOG_INFO("API", "Restart device");
delay(10); delay(10);
@@ -126,10 +104,12 @@ void NodeService::handleUpdateUpload(AsyncWebServerRequest* request, const Strin
} }
void NodeService::handleRestartRequest(AsyncWebServerRequest* request) { void NodeService::handleRestartRequest(AsyncWebServerRequest* request) {
AsyncWebServerResponse* response = request->beginResponse(200, "application/json", NodeOperationResponse response;
"{\"status\": \"restarting\"}"); response.setSuccess("restarting");
response->addHeader("Connection", "close");
request->send(response); AsyncWebServerResponse* httpResponse = request->beginResponse(200, "application/json", response.toJsonString());
httpResponse->addHeader("Connection", "close");
request->send(httpResponse);
request->onDisconnect([this]() { request->onDisconnect([this]() {
LOG_INFO("API", "Restart device"); LOG_INFO("API", "Restart device");
delay(10); delay(10);
@@ -138,36 +118,7 @@ void NodeService::handleRestartRequest(AsyncWebServerRequest* request) {
} }
void NodeService::handleEndpointsRequest(AsyncWebServerRequest* request) { void NodeService::handleEndpointsRequest(AsyncWebServerRequest* request) {
JsonDocument doc; NodeEndpointsResponse response;
JsonArray endpointsArr = doc["endpoints"].to<JsonArray>(); response.addEndpoints(apiServer.getEndpoints());
request->send(200, "application/json", response.toJsonString());
// Add all registered endpoints from ApiServer
for (const auto& endpoint : apiServer.getEndpoints()) {
JsonObject obj = endpointsArr.add<JsonObject>();
obj["uri"] = endpoint.uri;
obj["method"] = ApiServer::methodToStr(endpoint.method);
if (!endpoint.params.empty()) {
JsonArray paramsArr = obj["params"].to<JsonArray>();
for (const auto& ps : endpoint.params) {
JsonObject p = paramsArr.add<JsonObject>();
p["name"] = ps.name;
p["location"] = ps.location;
p["required"] = ps.required;
p["type"] = ps.type;
if (!ps.values.empty()) {
JsonArray allowed = p["values"].to<JsonArray>();
for (const auto& v : ps.values) {
allowed.add(v);
}
}
if (ps.defaultValue.length() > 0) {
p["default"] = ps.defaultValue;
}
}
}
}
String json;
serializeJson(doc, json);
request->send(200, "application/json", json);
} }

View File

@@ -1,13 +1,17 @@
#include "spore/services/TaskService.h" #include "spore/services/TaskService.h"
#include "spore/core/ApiServer.h" #include "spore/core/ApiServer.h"
#include "spore/types/TaskResponse.h"
#include <algorithm> #include <algorithm>
using spore::types::TaskStatusResponse;
using spore::types::TaskControlResponse;
TaskService::TaskService(TaskManager& taskManager) : taskManager(taskManager) {} TaskService::TaskService(TaskManager& taskManager) : taskManager(taskManager) {}
void TaskService::registerEndpoints(ApiServer& api) { void TaskService::registerEndpoints(ApiServer& api) {
api.addEndpoint("/api/tasks/status", HTTP_GET, api.addEndpoint("/api/tasks/status", HTTP_GET,
[this](AsyncWebServerRequest* request) { handleStatusRequest(request); }, [this](AsyncWebServerRequest* request) { handleStatusRequest(request); },
std::vector<ParamSpec>{}); std::vector<ParamSpec>{}, "TaskService");
api.addEndpoint("/api/tasks/control", HTTP_POST, api.addEndpoint("/api/tasks/control", HTTP_POST,
[this](AsyncWebServerRequest* request) { handleControlRequest(request); }, [this](AsyncWebServerRequest* request) { handleControlRequest(request); },
@@ -28,36 +32,19 @@ void TaskService::registerEndpoints(ApiServer& api) {
{String("enable"), String("disable"), String("start"), String("stop"), String("status")}, {String("enable"), String("disable"), String("start"), String("stop"), String("status")},
String("") String("")
} }
}); }, "TaskService");
} }
void TaskService::handleStatusRequest(AsyncWebServerRequest* request) { void TaskService::handleStatusRequest(AsyncWebServerRequest* request) {
TaskStatusResponse response;
// Get task statuses using a separate document to avoid reference issues
JsonDocument scratch; JsonDocument scratch;
auto taskStatuses = taskManager.getAllTaskStatuses(scratch); auto taskStatuses = taskManager.getAllTaskStatuses(scratch);
JsonDocument doc; // Build the complete response with the task data
JsonObject summaryObj = doc["summary"].to<JsonObject>(); response.buildCompleteResponse(taskStatuses);
summaryObj["totalTasks"] = taskStatuses.size(); request->send(200, "application/json", response.toJsonString());
summaryObj["activeTasks"] = std::count_if(taskStatuses.begin(), taskStatuses.end(),
[](const auto& pair) { return pair.second["enabled"]; });
JsonArray tasksArr = doc["tasks"].to<JsonArray>();
for (const auto& taskPair : taskStatuses) {
JsonObject taskObj = tasksArr.add<JsonObject>();
taskObj["name"] = taskPair.first;
taskObj["interval"] = taskPair.second["interval"];
taskObj["enabled"] = taskPair.second["enabled"];
taskObj["running"] = taskPair.second["running"];
taskObj["autoStart"] = taskPair.second["autoStart"];
}
JsonObject systemObj = doc["system"].to<JsonObject>();
systemObj["freeHeap"] = ESP.getFreeHeap();
systemObj["uptime"] = millis();
String json;
serializeJson(doc, json);
request->send(200, "application/json", json);
} }
void TaskService::handleControlRequest(AsyncWebServerRequest* request) { void TaskService::handleControlRequest(AsyncWebServerRequest* request) {
@@ -88,50 +75,27 @@ void TaskService::handleControlRequest(AsyncWebServerRequest* request) {
success = true; success = true;
message = "Task status retrieved"; message = "Task status retrieved";
JsonDocument statusDoc; TaskControlResponse response;
statusDoc["success"] = success; response.setResponse(success, message, taskName, action);
statusDoc["message"] = message; response.addTaskDetails(taskName,
statusDoc["task"] = taskName; taskManager.isTaskEnabled(taskName.c_str()),
statusDoc["action"] = action; taskManager.isTaskRunning(taskName.c_str()),
taskManager.getTaskInterval(taskName.c_str()));
statusDoc["taskDetails"] = JsonObject(); request->send(200, "application/json", response.toJsonString());
JsonObject taskDetails = statusDoc["taskDetails"];
taskDetails["name"] = taskName;
taskDetails["enabled"] = taskManager.isTaskEnabled(taskName.c_str());
taskDetails["running"] = taskManager.isTaskRunning(taskName.c_str());
taskDetails["interval"] = taskManager.getTaskInterval(taskName.c_str());
taskDetails["system"] = JsonObject();
JsonObject systemInfo = taskDetails["system"];
systemInfo["freeHeap"] = ESP.getFreeHeap();
systemInfo["uptime"] = millis();
String statusJson;
serializeJson(statusDoc, statusJson);
request->send(200, "application/json", statusJson);
return; return;
} else { } else {
success = false; success = false;
message = "Invalid action. Use: enable, disable, start, stop, or status"; message = "Invalid action. Use: enable, disable, start, stop, or status";
} }
JsonDocument doc; TaskControlResponse response;
doc["success"] = success; response.setResponse(success, message, taskName, action);
doc["message"] = message; request->send(success ? 200 : 400, "application/json", response.toJsonString());
doc["task"] = taskName;
doc["action"] = action;
String json;
serializeJson(doc, json);
request->send(success ? 200 : 400, "application/json", json);
} else { } else {
JsonDocument doc; TaskControlResponse response;
doc["success"] = false; response.setError("Missing parameters. Required: task, action",
doc["message"] = "Missing parameters. Required: task, action"; "{\"task\": \"discovery_send\", \"action\": \"status\"}");
doc["example"] = "{\"task\": \"discovery_send\", \"action\": \"status\"}"; request->send(400, "application/json", response.toJsonString());
String json;
serializeJson(doc, json);
request->send(400, "application/json", json);
} }
} }

View File

@@ -10,11 +10,10 @@ Config::Config() {
api_server_port = 80; api_server_port = 80;
// Cluster Configuration // Cluster Configuration
discovery_interval_ms = 1000; // TODO retire this in favor of heartbeat_interval_ms discovery_interval_ms = 1000;
cluster_listen_interval_ms = 10; heartbeat_interval_ms = 2000;
heartbeat_interval_ms = 5000;
status_update_interval_ms = 1000; status_update_interval_ms = 1000;
member_info_update_interval_ms = 10000; // TODO retire this in favor of heartbeat_interval_ms member_info_update_interval_ms = 10000;
print_interval_ms = 5000; print_interval_ms = 5000;
// Node Status Thresholds // Node Status Thresholds

1
test/.gitignore vendored
View File

@@ -1 +0,0 @@
node_modules/

View File

@@ -1,64 +1,11 @@
# Test Scripts
This directory contains JavaScript test scripts to interact with the Spore device, primarily for testing cluster event broadcasting. This directory is intended for PlatformIO Test Runner and project tests.
## Prerequisites Unit Testing is a software testing method by which individual units of
source code, sets of one or more MCU program modules together with associated
These scripts require [Node.js](https://nodejs.org/) to be installed on your system. control data, usage procedures, and operating procedures, are tested to
determine whether they are fit for use. Unit testing finds problems early
## How to Run in the development cycle.
### 1. HTTP Cluster Broadcast Color (`test/http-cluster-broadcast-color.js`)
This script sends HTTP POST requests to the `/api/cluster/event` endpoint on your Spore device. It broadcasts NeoPattern color changes across the cluster every 5 seconds.
**Usage:**
```
node test/http-cluster-broadcast-color.js <device-ip>
```
Example:
```
node test/http-cluster-broadcast-color.js 10.0.1.53
```
This will broadcast `{ event: "api/neopattern/color", data: { color: "#RRGGBB", brightness: 128 } }` every 5 seconds to the cluster via `/api/cluster/event`.
### 2. WS Local Color Setter (`test/ws-color-client.js`)
Connects to the device WebSocket (`/ws`) and sets a solid color locally (non-broadcast) every 5 seconds by firing `api/neopattern/color`.
**Usage:**
```
node test/ws-color-client.js ws://<device-ip>/ws
```
Example:
```
node test/ws-color-client.js ws://10.0.1.53/ws
```
### 3. WS Cluster Broadcast Color (`test/ws-cluster-broadcast-color.js`)
Connects to the device WebSocket (`/ws`) and broadcasts a color change to all peers every 5 seconds by firing `cluster/broadcast` with the proper envelope.
**Usage:**
```
node test/ws-cluster-broadcast-color.js ws://<device-ip>/ws
```
Example:
```
node test/ws-cluster-broadcast-color.js ws://10.0.1.53/ws
```
### 4. WS Cluster Broadcast Rainbow (`test/ws-cluster-broadcast-rainbow.js`)
Broadcasts a smooth rainbow color transition over WebSocket using `cluster/broadcast` and the `api/neopattern/color` event. Update rate defaults to `UPDATE_RATE` in the script (e.g., 100 ms).
**Usage:**
```
node test/ws-cluster-broadcast-rainbow.js ws://<device-ip>/ws
```
Example:
```
node test/ws-cluster-broadcast-rainbow.js ws://10.0.1.53/ws
```
Note: Very fast update intervals (e.g., 10 ms) may saturate links or the device.
More information about PlatformIO Unit Testing:
- https://docs.platformio.org/en/latest/advanced/unit-testing/index.html

View File

@@ -1,52 +0,0 @@
// Simple HTTP client to broadcast a neopattern color change to the cluster
// Usage: node cluster-broadcast-color.js 10.0.1.53
const http = require('http');
const host = process.argv[2] || '127.0.0.1';
const port = 80;
const colors = ['#FF0000', '#00FF00', '#0000FF', '#FFFF00', '#FF00FF', '#00FFFF'];
let idx = 0;
function postClusterEvent(event, payloadObj) {
const payload = encodeURIComponent(JSON.stringify(payloadObj));
const body = `event=${encodeURIComponent(event)}&payload=${payload}`;
const options = {
host,
port,
path: '/api/cluster/event',
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': Buffer.byteLength(body)
}
};
const req = http.request(options, (res) => {
let data = '';
res.on('data', (chunk) => (data += chunk));
res.on('end', () => {
console.log('Response:', res.statusCode, data);
});
});
req.on('error', (err) => {
console.error('Request error:', err.message);
});
req.write(body);
req.end();
}
console.log(`Broadcasting color changes to http://${host}/api/cluster/event ...`);
setInterval(() => {
const color = colors[idx % colors.length];
idx++;
const payload = { color, brightness: 128 };
console.log('Broadcasting color:', payload);
postClusterEvent('api/neopattern/color', payload);
}, 5000);

33
test/package-lock.json generated
View File

@@ -1,33 +0,0 @@
{
"name": "test",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"dependencies": {
"ws": "^8.18.3"
}
},
"node_modules/ws": {
"version": "8.18.3",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.18.3.tgz",
"integrity": "sha512-PEIGCY5tSlUt50cqyMXfCzX+oOPqN0vuGqWzbcJ2xvnkzkq46oOpz7dQaTDBdfICb4N14+GARUDw2XV2N4tvzg==",
"license": "MIT",
"engines": {
"node": ">=10.0.0"
},
"peerDependencies": {
"bufferutil": "^4.0.1",
"utf-8-validate": ">=5.0.2"
},
"peerDependenciesMeta": {
"bufferutil": {
"optional": true
},
"utf-8-validate": {
"optional": true
}
}
}
}
}

View File

@@ -1,5 +0,0 @@
{
"dependencies": {
"ws": "^8.18.3"
}
}

View File

@@ -1,46 +0,0 @@
// WebSocket client to broadcast neopattern color changes across the cluster
// Usage: node ws-cluster-broadcast-color.js ws://<device-ip>/ws
const WebSocket = require('ws');
const url = process.argv[2] || 'ws://127.0.0.1/ws';
const ws = new WebSocket(url);
const colors = ['#FF0000', '#00FF00', '#0000FF', '#FFFF00', '#FF00FF', '#00FFFF'];
let idx = 0;
ws.on('open', () => {
console.log('Connected to', url);
// Broadcast color change every 5 seconds via cluster/broadcast
setInterval(() => {
const color = colors[idx % colors.length];
idx++;
const payload = { color, brightness: 128 };
const envelope = {
event: 'api/neopattern/color',
data: payload // server will serialize object payloads
};
const msg = { event: 'cluster/broadcast', payload: envelope };
ws.send(JSON.stringify(msg));
console.log('Broadcasted color event', payload);
}, 5000);
});
ws.on('message', (data) => {
try {
const msg = JSON.parse(data.toString());
console.log('Received:', msg);
} catch (e) {
console.log('Received raw:', data.toString());
}
});
ws.on('error', (err) => {
console.error('WebSocket error:', err.message);
});
ws.on('close', () => {
console.log('WebSocket closed');
});

View File

@@ -1,71 +0,0 @@
// WebSocket client to broadcast smooth rainbow color changes across the cluster
// Usage: node ws-cluster-broadcast-rainbow.js ws://<device-ip>/ws
const WebSocket = require('ws');
const url = process.argv[2] || 'ws://127.0.0.1/ws';
const ws = new WebSocket(url);
function hsvToRgb(h, s, v) {
const c = v * s;
const x = c * (1 - Math.abs(((h / 60) % 2) - 1));
const m = v - c;
let r = 0, g = 0, b = 0;
if (h < 60) { r = c; g = x; b = 0; }
else if (h < 120) { r = x; g = c; b = 0; }
else if (h < 180) { r = 0; g = c; b = x; }
else if (h < 240) { r = 0; g = x; b = c; }
else if (h < 300) { r = x; g = 0; b = c; }
else { r = c; g = 0; b = x; }
const R = Math.round((r + m) * 255);
const G = Math.round((g + m) * 255);
const B = Math.round((b + m) * 255);
return { r: R, g: G, b: B };
}
function toHex({ r, g, b }) {
const h = (n) => n.toString(16).padStart(2, '0').toUpperCase();
return `#${h(r)}${h(g)}${h(b)}`;
}
let hue = 0;
const SAT = 1.0; // full saturation
const VAL = 1.0; // full value
const BRIGHTNESS = 128;
const UPDATE_RATE = 100; // 100 ms
let timer = null;
ws.on('open', () => {
console.log('Connected to', url);
// UPDATE_RATE ms updates (10 Hz). Be aware this can saturate slow links.
timer = setInterval(() => {
const rgb = hsvToRgb(hue, SAT, VAL);
const color = toHex(rgb);
const envelope = {
event: 'api/neopattern/color',
data: { color, brightness: BRIGHTNESS }
};
const msg = { event: 'cluster/broadcast', payload: envelope };
try {
ws.send(JSON.stringify(msg));
} catch (_) {}
hue = (hue + 2) % 360; // advance hue (adjust for speed)
}, UPDATE_RATE);
});
ws.on('message', (data) => {
// Optionally throttle logs: comment out for quieter output
// console.log('WS:', data.toString());
});
ws.on('error', (err) => {
console.error('WebSocket error:', err.message);
});
ws.on('close', () => {
if (timer) clearInterval(timer);
console.log('WebSocket closed');
});

View File

@@ -1,48 +0,0 @@
// Simple WebSocket client to test streaming API color changes
// Usage: node ws-color-client.js ws://<device-ip>/ws
const WebSocket = require('ws');
const url = process.argv[2] || 'ws://127.0.0.1/ws';
const ws = new WebSocket(url);
const colors = [
'#FF0000', // red
'#00FF00', // green
'#0000FF', // blue
'#FFFF00', // yellow
'#FF00FF', // magenta
'#00FFFF' // cyan
];
let idx = 0;
ws.on('open', () => {
console.log('Connected to', url);
// Send a message every 5 seconds to set solid color
setInterval(() => {
const color = colors[idx % colors.length];
idx++;
const payload = { color, brightness: 128 };
// Send payload as an object (server supports string or object)
const msg = { event: 'api/neopattern/color', payload };
ws.send(JSON.stringify(msg));
console.log('Sent color event', payload);
}, 5000);
});
ws.on('message', (data) => {
try {
const msg = JSON.parse(data.toString());
console.log('Received:', msg);
} catch (e) {
console.log('Received raw:', data.toString());
}
});
ws.on('error', (err) => {
console.error('WebSocket error:', err.message);
});
ws.on('close', () => {
console.log('WebSocket closed');
});