mirror of
https://github.com/rstrouse/ESPSomfy-RTS.git
synced 2026-10-31 13:52:15 +01:00
fix: protocol 2 core review findings
From the review of #99: - A sync could stall when a client took a slot the moment another left: loop() used up the new client's sync on the old one. The slots move to ProtoSlots, a pure module with native tests, and loop() takes a slot's wants only for the client it read (takeWants). - Deleting a group sends its shades, whose inGroup changed. Deleting a shade unlinks it from its groups and sends them, so a new shade given its id doesn't join them. - next and sync with an id get an ok reply, and a message in pieces gets one bad request, not one per piece. - Feeds check for loop() before they write the event buffer, and log an event too large, as events do. - editChanges() (ModelEditor, native tests) says what clients get after an edit; the edit result carries the shade a group gained or lost. order() takes a sort_list_t. - The frame writers move to ProtoFrame. Renames: pf to proto_field_t, the X-macro letters to TYPE/FIELD/ERR/FEED, proto_request_t to proto_message_t, REQ_* to WANTS_*, fmt to eventJson. The event bodies share sendItem(), the connection type name lives in ESPNetwork, and the loop() task check in SocketEmitter. - e2e: the heap floor check skips when the low since boot is already under it; tests for replies to next and sync and a long message. - SECURITY_PLAN.md splits step 1 into 1a-1d. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
cjkas
parent
d869bb53d5
commit
dc2f1478b6
29 changed files with 1087 additions and 670 deletions
|
|
@ -1,10 +1,12 @@
|
|||
#include <Arduino.h>
|
||||
#include <WiFi.h>
|
||||
#include <atomic>
|
||||
#include <algorithm>
|
||||
#include "esp_log.h"
|
||||
#include "ProtoSocket.h"
|
||||
#include "ProtoSync.h"
|
||||
#include "ProtoFrame.h"
|
||||
#include "ProtoParse.h"
|
||||
#include "ProtoSlots.h"
|
||||
#include "SomfySnapshot.h"
|
||||
#include "Somfy.h"
|
||||
#include "ConfigSettings.h"
|
||||
|
|
@ -23,87 +25,62 @@ extern ConfigSettings settings;
|
|||
extern ESPNetwork net;
|
||||
extern Web webServer;
|
||||
extern LoopStats loopStats;
|
||||
extern SocketEmitter sockEmit;
|
||||
|
||||
ProtoSocket protoSocket;
|
||||
static AsyncWebSocket ws("/ws");
|
||||
static_assert(MAX_SOCK_RESPONSE >= PROTO_MAX_FRAME, "protocol 2 frames share the socket event buffer");
|
||||
|
||||
// What a client asked for, from async_tcp to loop().
|
||||
enum : uint32_t {
|
||||
REQ_START = 0x01, // just connected
|
||||
REQ_NEXT = 0x02,
|
||||
REQ_RESYNC = 0x04
|
||||
};
|
||||
struct proto_slot_t {
|
||||
// Set and cleared on async_tcp; 0 when the slot is free.
|
||||
std::atomic<uint32_t> asyncId{0};
|
||||
std::atomic<uint32_t> requests{0};
|
||||
std::atomic<uint8_t> feeds{0}; // a bit per feed number
|
||||
std::atomic<bool> withSchema{true};
|
||||
// loop() only.
|
||||
uint32_t loopId = 0; // the client loop() is serving here
|
||||
sync_cursor_t cur;
|
||||
bool awaitingNext = false;
|
||||
uint32_t sentAt = 0;
|
||||
};
|
||||
static proto_slot_t slots[PROTO_MAX_CLIENTS];
|
||||
static BufferedJsonFormatter fmt;
|
||||
|
||||
static proto_slot_t *slotOf(uint32_t asyncId) {
|
||||
for(proto_slot_t &s : slots)
|
||||
if(s.asyncId.load() == asyncId) return &s;
|
||||
return nullptr;
|
||||
}
|
||||
// What loop() formats events and syncEnd with, into the socket event buffer.
|
||||
static BufferedJsonFormatter eventJson;
|
||||
|
||||
/*********************************************************************
|
||||
* The shade model as a sync source
|
||||
********************************************************************/
|
||||
static void writeHeader(ProtoWriter &w) {
|
||||
w.add(pf::maxRooms, (uint8_t)SOMFY_MAX_ROOMS);
|
||||
w.add(pf::maxShades, (uint8_t)SOMFY_MAX_SHADES);
|
||||
w.add(pf::maxGroups, (uint8_t)SOMFY_MAX_GROUPS);
|
||||
w.add(pf::maxGroupedShades, (uint8_t)SOMFY_MAX_GROUPED_SHADES);
|
||||
w.add(pf::maxLinkedRemotes, (uint8_t)SOMFY_MAX_LINKED_REMOTES);
|
||||
w.add(pf::startingAddress, somfy.startingAddress);
|
||||
w.add(proto_field_t::maxRooms, (uint8_t)SOMFY_MAX_ROOMS);
|
||||
w.add(proto_field_t::maxShades, (uint8_t)SOMFY_MAX_SHADES);
|
||||
w.add(proto_field_t::maxGroups, (uint8_t)SOMFY_MAX_GROUPS);
|
||||
w.add(proto_field_t::maxGroupedShades, (uint8_t)SOMFY_MAX_GROUPED_SHADES);
|
||||
w.add(proto_field_t::maxLinkedRemotes, (uint8_t)SOMFY_MAX_LINKED_REMOTES);
|
||||
w.add(proto_field_t::startingAddress, somfy.startingAddress);
|
||||
// What a new shade or group gets.
|
||||
w.add(pf::bitLength, somfy.transceiver.config.type);
|
||||
w.add(pf::proto, static_cast<uint8_t>(somfy.transceiver.config.proto));
|
||||
w.add(pf::serverId, settings.serverId);
|
||||
w.add(pf::hostname, settings.hostname);
|
||||
w.add(pf::model, "ESPSomfyRTS");
|
||||
w.add(pf::chipModel, settings.chipModel);
|
||||
w.add(pf::fw, settings.fwVersion.name);
|
||||
const char *connType = "Unknown";
|
||||
if(net.connType == conn_types_t::ethernet) connType = "Ethernet";
|
||||
else if(net.connType == conn_types_t::wifi) connType = "Wifi";
|
||||
w.add(pf::connType, connType);
|
||||
w.add(proto_field_t::bitLength, somfy.transceiver.config.type);
|
||||
w.add(proto_field_t::proto, static_cast<uint8_t>(somfy.transceiver.config.proto));
|
||||
w.add(proto_field_t::serverId, settings.serverId);
|
||||
w.add(proto_field_t::hostname, settings.hostname);
|
||||
w.add(proto_field_t::model, "ESPSomfyRTS");
|
||||
w.add(proto_field_t::chipModel, settings.chipModel);
|
||||
w.add(proto_field_t::fw, settings.fwVersion.name);
|
||||
w.add(proto_field_t::connType, net.connTypeName());
|
||||
}
|
||||
static void writeMemory(ProtoWriter &w) {
|
||||
w.add(pf::freeHeap, ESP.getFreeHeap());
|
||||
w.add(pf::maxHeap, ESP.getMaxAllocHeap());
|
||||
w.add(pf::minHeap, ESP.getMinFreeHeap());
|
||||
w.add(pf::totalHeap, ESP.getHeapSize());
|
||||
w.add(pf::uptime, (uint32_t)millis());
|
||||
w.add(pf::loopRate, loopStats.publishedRate());
|
||||
w.add(pf::loopMaxUs, loopStats.publishedMaxUs());
|
||||
w.add(proto_field_t::freeHeap, ESP.getFreeHeap());
|
||||
w.add(proto_field_t::maxHeap, ESP.getMaxAllocHeap());
|
||||
w.add(proto_field_t::minHeap, ESP.getMinFreeHeap());
|
||||
w.add(proto_field_t::totalHeap, ESP.getHeapSize());
|
||||
w.add(proto_field_t::uptime, (uint32_t)millis());
|
||||
w.add(proto_field_t::loopRate, loopStats.publishedRate());
|
||||
w.add(proto_field_t::loopMaxUs, loopStats.publishedMaxUs());
|
||||
}
|
||||
static void writeWifi(ProtoWriter &w) {
|
||||
if(WiFi.status() == WL_CONNECTED) {
|
||||
w.add(pf::ssid, WiFi.SSID().c_str());
|
||||
w.add(pf::strength, (int32_t)WiFi.RSSI());
|
||||
w.add(pf::channel, (int32_t)net.channel);
|
||||
w.add(proto_field_t::ssid, WiFi.SSID().c_str());
|
||||
w.add(proto_field_t::strength, (int32_t)WiFi.RSSI());
|
||||
w.add(proto_field_t::channel, (int32_t)net.channel);
|
||||
}
|
||||
else {
|
||||
w.add(pf::strength, (int32_t)-100);
|
||||
w.add(pf::channel, (int32_t)-1);
|
||||
w.add(proto_field_t::strength, (int32_t)-100);
|
||||
w.add(proto_field_t::channel, (int32_t)-1);
|
||||
}
|
||||
}
|
||||
static void writeEthernet(ProtoWriter &w) {
|
||||
#if HAS_ETHERNET
|
||||
if(net.connType == conn_types_t::ethernet) {
|
||||
w.add(pf::connected, net.connected());
|
||||
w.add(pf::speed, (uint32_t)ETH.linkSpeed());
|
||||
w.add(pf::fullDuplex, ETH.fullDuplex());
|
||||
w.add(proto_field_t::connected, net.connected());
|
||||
w.add(proto_field_t::speed, (uint32_t)ETH.linkSpeed());
|
||||
w.add(proto_field_t::fullDuplex, ETH.fullDuplex());
|
||||
}
|
||||
#else
|
||||
(void)w;
|
||||
|
|
@ -162,13 +139,10 @@ static void replyError(AsyncWebSocketClient *client, uint32_t id, proto_error_t
|
|||
writeErrorReply(json, e.id, e.code, e.desc);
|
||||
}, &e);
|
||||
}
|
||||
// Only a message with an id wants a reply.
|
||||
static void replyOk(AsyncWebSocketClient *client, uint32_t id) {
|
||||
if(id == 0) return;
|
||||
sendFixed(client, [](JsonFormatter &json, void *ctx) {
|
||||
beginFrame(json, proto_type_t::reply, *static_cast<uint32_t *>(ctx));
|
||||
ProtoWriter(json).add(pf::ok, true);
|
||||
endFrame(json);
|
||||
}, &id);
|
||||
sendFixed(client, [](JsonFormatter &json, void *ctx) { writeOkReply(json, *static_cast<uint32_t *>(ctx)); }, &id);
|
||||
}
|
||||
static bool authorized(AsyncWebServerRequest *request) {
|
||||
if(settings.Security.type == security_types::None) return true;
|
||||
|
|
@ -180,68 +154,65 @@ static bool authorized(AsyncWebServerRequest *request) {
|
|||
}
|
||||
static void onConnect(AsyncWebSocketClient *client, AsyncWebServerRequest *request) {
|
||||
const uint32_t id = client->id();
|
||||
proto_slot_t *slot = nullptr;
|
||||
for(proto_slot_t &s : slots) {
|
||||
uint32_t expected = 0;
|
||||
if(s.asyncId.compare_exchange_strong(expected, id)) { slot = &s; break; }
|
||||
}
|
||||
const AsyncWebParameter *schema = request ? request->getParam("schema") : nullptr;
|
||||
const bool withSchema = !(schema && schema->value() == protoSchemaHash());
|
||||
proto_slot_t *slot = claimSlot(slots, PROTO_MAX_CLIENTS, id, withSchema);
|
||||
if(!slot) {
|
||||
ESP_LOGW(TAG, "Socket [%lu] refused: all %u clients connected", (unsigned long)id, PROTO_MAX_CLIENTS);
|
||||
client->close();
|
||||
return;
|
||||
}
|
||||
const AsyncWebParameter *schema = request ? request->getParam("schema") : nullptr;
|
||||
slot->withSchema = !(schema && schema->value() == protoSchemaHash());
|
||||
slot->feeds = 0;
|
||||
client->setCloseClientOnQueueFull(true);
|
||||
sendFixed(client, [](JsonFormatter &json, void *) { writeHello(json, settings.fwVersion.name); }, nullptr);
|
||||
slot->requests.fetch_or(REQ_START);
|
||||
startSync(*slot);
|
||||
IPAddress ip = client->remoteIP();
|
||||
ESP_LOGI(TAG, "Socket [%lu] connected from %u.%u.%u.%u%s", (unsigned long)id, ip[0], ip[1], ip[2], ip[3], slot->withSchema ? "" : " (has the schema)");
|
||||
ESP_LOGI(TAG, "Socket [%lu] connected from %u.%u.%u.%u%s", (unsigned long)id, ip[0], ip[1], ip[2], ip[3], withSchema ? "" : " (has the schema)");
|
||||
}
|
||||
static void onDisconnect(AsyncWebSocketClient *client) {
|
||||
proto_slot_t *slot = slotOf(client->id());
|
||||
proto_slot_t *slot = slotOf(slots, PROTO_MAX_CLIENTS, client->id());
|
||||
if(!slot) return;
|
||||
slot->requests = 0;
|
||||
slot->feeds = 0;
|
||||
slot->asyncId = 0;
|
||||
freeSlot(*slot);
|
||||
ESP_LOGI(TAG, "Socket [%lu] disconnected", (unsigned long)client->id());
|
||||
}
|
||||
static void onFrame(AsyncWebSocketClient *client, AwsFrameInfo *info, uint8_t *data, size_t len) {
|
||||
if(!(info->final && info->index == 0 && info->len == len && info->opcode == WS_TEXT)) {
|
||||
replyError(client, 0, proto_error_t::badRequest, "Send each message as one text frame");
|
||||
// One reply for the whole message: its first piece gets it, the rest are
|
||||
// dropped. A frame larger than a TCP segment arrives in pieces too.
|
||||
if(info->num == 0 && info->index == 0) replyError(client, 0, proto_error_t::badRequest, "Send each message as one text frame");
|
||||
return;
|
||||
}
|
||||
proto_slot_t *slot = slotOf(client->id());
|
||||
proto_slot_t *slot = slotOf(slots, PROTO_MAX_CLIENTS, client->id());
|
||||
if(!slot) return;
|
||||
JsonDocument doc;
|
||||
proto_request_t req;
|
||||
const proto_error_t err = parseClientFrame(reinterpret_cast<const char *>(data), len, doc, req);
|
||||
proto_message_t msg;
|
||||
const proto_error_t err = parseClientFrame(reinterpret_cast<const char *>(data), len, doc, msg);
|
||||
if(err != proto_error_t::none) {
|
||||
replyError(client, req.id, err, err == proto_error_t::unknownOp ? "Unknown message type" : "Malformed message");
|
||||
replyError(client, msg.id, err, err == proto_error_t::unknownOp ? "Unknown message type" : "Malformed message");
|
||||
return;
|
||||
}
|
||||
switch(static_cast<proto_type_t>(req.type)) {
|
||||
switch(static_cast<proto_type_t>(msg.type)) {
|
||||
case proto_type_t::next:
|
||||
slot->requests.fetch_or(REQ_NEXT);
|
||||
slot->wants.fetch_or(WANTS_NEXT);
|
||||
replyOk(client, msg.id);
|
||||
return;
|
||||
case proto_type_t::sync:
|
||||
slot->requests.fetch_or(REQ_RESYNC);
|
||||
slot->wants.fetch_or(WANTS_SYNC_AGAIN);
|
||||
replyOk(client, msg.id);
|
||||
return;
|
||||
case proto_type_t::subscribe:
|
||||
case proto_type_t::unsubscribe: {
|
||||
const int32_t feed = protoInt(doc["d"], pf::feed);
|
||||
const int32_t feed = protoInt(doc["d"], proto_field_t::feed);
|
||||
bool known = false;
|
||||
for(uint8_t i = 0; i < protoFeedCount(); i++) known |= protoFeedAt(i).number == feed;
|
||||
if(!known) { replyError(client, req.id, proto_error_t::badRequest, "Unknown feed"); return; }
|
||||
if(req.type == static_cast<uint8_t>(proto_type_t::subscribe)) slot->feeds.fetch_or(1 << feed);
|
||||
if(!known) { replyError(client, msg.id, proto_error_t::badRequest, "Unknown feed"); return; }
|
||||
if(msg.type == static_cast<uint8_t>(proto_type_t::subscribe)) slot->feeds.fetch_or(1 << feed);
|
||||
else slot->feeds.fetch_and(~(1 << feed));
|
||||
replyOk(client, req.id);
|
||||
replyOk(client, msg.id);
|
||||
return;
|
||||
}
|
||||
default:
|
||||
// The ops arrive with the protocol core's later PRs.
|
||||
replyError(client, req.id, proto_error_t::unknownOp, "Unknown message type");
|
||||
replyError(client, msg.id, proto_error_t::unknownOp, "Unknown message type");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
|
@ -256,17 +227,16 @@ void ProtoSocket::wsEvent(AsyncWebSocket *server, AsyncWebSocketClient *client,
|
|||
}
|
||||
|
||||
void ProtoSocket::begin(AsyncWebServer &server) {
|
||||
this->_owner = xTaskGetCurrentTaskHandle();
|
||||
ws.handleHandshake(authorized);
|
||||
ws.onEvent(ProtoSocket::wsEvent);
|
||||
server.addHandler(&ws);
|
||||
}
|
||||
bool ProtoSocket::onLoopTask() const { return emitAllowed(xTaskGetCurrentTaskHandle(), this->_owner); }
|
||||
bool ProtoSocket::hasClients() const {
|
||||
for(const proto_slot_t &s : slots)
|
||||
if(s.asyncId.load()) return true;
|
||||
if(s.clientId.load()) return true;
|
||||
return false;
|
||||
}
|
||||
bool ProtoSocket::sending() const { return sockEmit.onLoopTask() && this->hasClients(); }
|
||||
|
||||
/*********************************************************************
|
||||
* loop(): syncs
|
||||
|
|
@ -276,39 +246,26 @@ static void sendSyncFrame(proto_slot_t &s) {
|
|||
char *buf = socketFrameBuffer();
|
||||
const size_t len = writeSyncFrame(src, s.cur, buf, PROTO_MAX_FRAME);
|
||||
if(len) ws.text(s.loopId, buf, len);
|
||||
if(s.cur.active()) {
|
||||
s.awaitingNext = true;
|
||||
s.sentAt = millis();
|
||||
return;
|
||||
}
|
||||
fmt.setBuffer(buf, PROTO_MAX_FRAME);
|
||||
beginFrame(fmt, proto_type_t::syncEnd);
|
||||
endFrame(fmt);
|
||||
ws.text(s.loopId, buf, fmt.length());
|
||||
s.awaitingNext = false;
|
||||
syncFrameSent(s, millis());
|
||||
if(s.cur.active()) return;
|
||||
eventJson.setBuffer(buf, PROTO_MAX_FRAME);
|
||||
beginFrame(eventJson, proto_type_t::syncEnd);
|
||||
endFrame(eventJson);
|
||||
ws.text(s.loopId, buf, eventJson.length());
|
||||
}
|
||||
void ProtoSocket::loop() {
|
||||
if(!this->onLoopTask()) return;
|
||||
if(!sockEmit.onLoopTask()) return;
|
||||
for(proto_slot_t &s : slots) {
|
||||
const uint32_t id = s.asyncId.load();
|
||||
if(id != s.loopId) {
|
||||
// A client left or another took the slot.
|
||||
s.loopId = id;
|
||||
s.cur = sync_cursor_t();
|
||||
s.awaitingNext = false;
|
||||
}
|
||||
if(!id) continue;
|
||||
const uint32_t req = s.requests.exchange(0);
|
||||
if(req & (REQ_START | REQ_RESYNC)) {
|
||||
s.cur = beginSync((req & REQ_START) && s.withSchema.load());
|
||||
sendSyncFrame(s);
|
||||
}
|
||||
else if((req & REQ_NEXT) && s.awaitingNext) sendSyncFrame(s);
|
||||
else if(s.awaitingNext && millis() - s.sentAt > PROTO_SYNC_TIMEOUT_MS) {
|
||||
ESP_LOGW(TAG, "Socket [%lu] sent no next for %us, closing it", (unsigned long)id, PROTO_SYNC_TIMEOUT_MS / 1000);
|
||||
s.awaitingNext = false;
|
||||
s.cur = sync_cursor_t();
|
||||
ws.close(id);
|
||||
switch(nextSyncStep(s, millis())) {
|
||||
case sync_step_t::send:
|
||||
sendSyncFrame(s);
|
||||
break;
|
||||
case sync_step_t::timedOut:
|
||||
ESP_LOGW(TAG, "Socket [%lu] sent no next for %us, closing it", (unsigned long)s.loopId, PROTO_SYNC_TIMEOUT_MS / 1000);
|
||||
ws.close(s.loopId);
|
||||
break;
|
||||
case sync_step_t::none:
|
||||
break;
|
||||
}
|
||||
}
|
||||
ws.cleanupClients(PROTO_MAX_CLIENTS);
|
||||
|
|
@ -317,154 +274,148 @@ void ProtoSocket::loop() {
|
|||
/*********************************************************************
|
||||
* loop(): events
|
||||
********************************************************************/
|
||||
JsonFormatter &ProtoSocket::beginEvent(proto_type_t type) {
|
||||
fmt.setBuffer(socketFrameBuffer(), PROTO_MAX_FRAME);
|
||||
beginFrame(fmt, type);
|
||||
return fmt;
|
||||
static JsonFormatter &beginEvent(proto_type_t type) {
|
||||
eventJson.setBuffer(socketFrameBuffer(), PROTO_MAX_FRAME);
|
||||
beginFrame(eventJson, type);
|
||||
return eventJson;
|
||||
}
|
||||
void ProtoSocket::endEvent() {
|
||||
endFrame(fmt);
|
||||
if(fmt.overflowed()) {
|
||||
ESP_LOGE(TAG, "Event too large for %u bytes, not sent", PROTO_MAX_FRAME);
|
||||
return;
|
||||
}
|
||||
ws.textAll(socketFrameBuffer(), fmt.length());
|
||||
}
|
||||
// Each event checks the task and the clients first, so nothing is formatted
|
||||
// for nobody.
|
||||
#define PROTO_EVENT_GUARD() if(!this->onLoopTask() || !this->hasClients()) return
|
||||
|
||||
void ProtoSocket::shadeState(SomfyShade &shade) {
|
||||
PROTO_EVENT_GUARD();
|
||||
shade_snapshot_t s;
|
||||
snapshotShade(shade, s);
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::shadeState));
|
||||
protoShadeState(w, s);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::shade(SomfyShade &shade) {
|
||||
PROTO_EVENT_GUARD();
|
||||
shade_snapshot_t s;
|
||||
snapshotShade(shade, s);
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::shade));
|
||||
protoShade(w, s);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::shadeRemoved(uint8_t shadeId) {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter(this->beginEvent(proto_type_t::shadeRemoved)).add(pf::shadeId, shadeId);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::group(SomfyGroup &group) {
|
||||
PROTO_EVENT_GUARD();
|
||||
group_snapshot_t g;
|
||||
snapshotGroup(group, g);
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::group));
|
||||
protoGroup(w, g);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::groupState(SomfyGroup &group) {
|
||||
PROTO_EVENT_GUARD();
|
||||
group_snapshot_t g;
|
||||
snapshotGroup(group, g);
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::groupState));
|
||||
protoGroupState(w, g.groupId, g.flags, group.direction);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::groupRemoved(uint8_t groupId) {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter(this->beginEvent(proto_type_t::groupRemoved)).add(pf::groupId, groupId);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::room(SomfyRoom &room) {
|
||||
PROTO_EVENT_GUARD();
|
||||
room_snapshot_t r;
|
||||
snapshotRoom(room, r);
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::room));
|
||||
protoRoom(w, r);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::roomRemoved(uint8_t roomId) {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter(this->beginEvent(proto_type_t::roomRemoved)).add(pf::roomId, roomId);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::order(proto_type_t type) {
|
||||
PROTO_EVENT_GUARD();
|
||||
struct entry_t { int8_t order; uint8_t id; };
|
||||
entry_t items[SOMFY_MAX_SHADES];
|
||||
uint8_t n = 0;
|
||||
if(type == proto_type_t::roomOrder) {
|
||||
for(SomfyRoom &r : somfy.rooms) if(r.roomId != 0) items[n++] = {r.sortOrder, r.roomId};
|
||||
}
|
||||
else if(type == proto_type_t::shadeOrder) {
|
||||
for(SomfyShade &s : somfy.shades) if(s.getShadeId() != 255) items[n++] = {s.sortOrder, s.getShadeId()};
|
||||
}
|
||||
else {
|
||||
for(SomfyGroup &g : somfy.groups) if(g.getGroupId() != 255) items[n++] = {g.sortOrder, g.getGroupId()};
|
||||
}
|
||||
for(uint8_t i = 1; i < n; i++)
|
||||
for(uint8_t j = i; j > 0 && items[j].order < items[j - 1].order; j--) {
|
||||
const entry_t t = items[j];
|
||||
items[j] = items[j - 1];
|
||||
items[j - 1] = t;
|
||||
}
|
||||
ProtoWriter w(this->beginEvent(type));
|
||||
w.beginList(pf::ids);
|
||||
for(uint8_t i = 0; i < n; i++) w.item(items[i].id);
|
||||
w.endList();
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::repeaters() {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::repeaters));
|
||||
w.beginList(pf::repeaters);
|
||||
for(uint32_t address : somfy.repeaters) if(address != 0) w.item(address);
|
||||
w.endList();
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::shadeCommand(SomfyShade &shade, const char *command, const char *source, uint32_t sourceAddress) {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::shadeCommand));
|
||||
w.add(pf::shadeId, shade.getShadeId());
|
||||
w.add(pf::remoteAddress, shade.getRemoteAddress());
|
||||
w.add(pf::command, command);
|
||||
w.add(pf::source, source);
|
||||
w.add(pf::lastRollingCode, shade.lastRollingCode);
|
||||
w.add(pf::sourceAddress, sourceAddress);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::memory() {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::memory));
|
||||
writeMemory(w);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::wifi() {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::wifi));
|
||||
writeWifi(w);
|
||||
this->endEvent();
|
||||
}
|
||||
void ProtoSocket::ethernet() {
|
||||
PROTO_EVENT_GUARD();
|
||||
ProtoWriter w(this->beginEvent(proto_type_t::ethernet));
|
||||
writeEthernet(w);
|
||||
this->endEvent();
|
||||
}
|
||||
bool ProtoSocket::subscribed(proto_feed_t feed) const {
|
||||
const uint8_t bit = 1 << static_cast<uint8_t>(feed);
|
||||
for(const proto_slot_t &s : slots)
|
||||
if(s.asyncId.load() && (s.feeds.load() & bit)) return true;
|
||||
// Closes the event; false when it didn't fit, which is logged.
|
||||
static bool finishEvent() {
|
||||
endFrame(eventJson);
|
||||
if(!eventJson.overflowed()) return true;
|
||||
ESP_LOGE(TAG, "Event too large for %u bytes, not sent", PROTO_MAX_FRAME);
|
||||
return false;
|
||||
}
|
||||
ProtoWriter ProtoSocket::beginFeed(proto_type_t type) { return ProtoWriter(this->beginEvent(type)); }
|
||||
static void sendEvent() {
|
||||
if(finishEvent()) ws.textAll(socketFrameBuffer(), eventJson.length());
|
||||
}
|
||||
// An item whole, or only its state: a snapshot of it, written by write.
|
||||
template<typename Model, typename Snapshot>
|
||||
static void sendItem(proto_type_t type, Model &item, void (*snapshot)(Model &, Snapshot &), void (*write)(ProtoWriter &, const Snapshot &)) {
|
||||
Snapshot s;
|
||||
snapshot(item, s);
|
||||
ProtoWriter w(beginEvent(type));
|
||||
write(w, s);
|
||||
sendEvent();
|
||||
}
|
||||
|
||||
void ProtoSocket::shadeState(SomfyShade &shade) {
|
||||
if(this->sending()) sendItem(proto_type_t::shadeState, shade, snapshotShade, protoShadeState);
|
||||
}
|
||||
void ProtoSocket::shade(SomfyShade &shade) {
|
||||
if(this->sending()) sendItem(proto_type_t::shade, shade, snapshotShade, protoShade);
|
||||
}
|
||||
void ProtoSocket::shadeRemoved(uint8_t shadeId) {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter(beginEvent(proto_type_t::shadeRemoved)).add(proto_field_t::shadeId, shadeId);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::group(SomfyGroup &group) {
|
||||
if(this->sending()) sendItem(proto_type_t::group, group, snapshotGroup, protoGroup);
|
||||
}
|
||||
void ProtoSocket::groupState(SomfyGroup &group) {
|
||||
if(!this->sending()) return;
|
||||
group_snapshot_t g;
|
||||
snapshotGroup(group, g);
|
||||
ProtoWriter w(beginEvent(proto_type_t::groupState));
|
||||
protoGroupState(w, g.groupId, g.flags, group.direction);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::groupRemoved(uint8_t groupId) {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter(beginEvent(proto_type_t::groupRemoved)).add(proto_field_t::groupId, groupId);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::room(SomfyRoom &room) {
|
||||
if(this->sending()) sendItem(proto_type_t::room, room, snapshotRoom, protoRoom);
|
||||
}
|
||||
void ProtoSocket::roomRemoved(uint8_t roomId) {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter(beginEvent(proto_type_t::roomRemoved)).add(proto_field_t::roomId, roomId);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::order(sort_list_t list) {
|
||||
if(!this->sending()) return;
|
||||
struct entry_t { int8_t order; uint8_t id; };
|
||||
static constexpr uint8_t MOST = std::max({SOMFY_MAX_ROOMS, SOMFY_MAX_SHADES, SOMFY_MAX_GROUPS});
|
||||
entry_t items[MOST];
|
||||
uint8_t n = 0;
|
||||
proto_type_t type;
|
||||
switch(list) {
|
||||
case sort_list_t::rooms:
|
||||
type = proto_type_t::roomOrder;
|
||||
for(SomfyRoom &r : somfy.rooms) if(r.roomId != 0) items[n++] = {r.sortOrder, r.roomId};
|
||||
break;
|
||||
case sort_list_t::shades:
|
||||
type = proto_type_t::shadeOrder;
|
||||
for(SomfyShade &s : somfy.shades) if(s.getShadeId() != 255) items[n++] = {s.sortOrder, s.getShadeId()};
|
||||
break;
|
||||
case sort_list_t::groups:
|
||||
type = proto_type_t::groupOrder;
|
||||
for(SomfyGroup &g : somfy.groups) if(g.getGroupId() != 255) items[n++] = {g.sortOrder, g.getGroupId()};
|
||||
break;
|
||||
default:
|
||||
return;
|
||||
}
|
||||
for(uint8_t i = 1; i < n; i++)
|
||||
for(uint8_t j = i; j > 0 && items[j].order < items[j - 1].order; j--) std::swap(items[j], items[j - 1]);
|
||||
ProtoWriter w(beginEvent(type));
|
||||
w.beginList(proto_field_t::ids);
|
||||
for(uint8_t i = 0; i < n; i++) w.item(items[i].id);
|
||||
w.endList();
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::repeaters() {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter w(beginEvent(proto_type_t::repeaters));
|
||||
w.beginList(proto_field_t::repeaters);
|
||||
for(uint32_t address : somfy.repeaters) if(address != 0) w.item(address);
|
||||
w.endList();
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::shadeCommand(SomfyShade &shade, const char *command, const char *source, uint32_t sourceAddress) {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter w(beginEvent(proto_type_t::shadeCommand));
|
||||
w.add(proto_field_t::shadeId, shade.getShadeId());
|
||||
w.add(proto_field_t::remoteAddress, shade.getRemoteAddress());
|
||||
w.add(proto_field_t::command, command);
|
||||
w.add(proto_field_t::source, source);
|
||||
w.add(proto_field_t::lastRollingCode, shade.lastRollingCode);
|
||||
w.add(proto_field_t::sourceAddress, sourceAddress);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::memory() {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter w(beginEvent(proto_type_t::memory));
|
||||
writeMemory(w);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::wifi() {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter w(beginEvent(proto_type_t::wifi));
|
||||
writeWifi(w);
|
||||
sendEvent();
|
||||
}
|
||||
void ProtoSocket::ethernet() {
|
||||
if(!this->sending()) return;
|
||||
ProtoWriter w(beginEvent(proto_type_t::ethernet));
|
||||
writeEthernet(w);
|
||||
sendEvent();
|
||||
}
|
||||
bool ProtoSocket::subscribed(proto_feed_t feed) const {
|
||||
// The feed is formatted into the event buffer, which is loop()'s.
|
||||
if(!sockEmit.onLoopTask()) return false;
|
||||
const uint8_t bit = 1 << static_cast<uint8_t>(feed);
|
||||
for(const proto_slot_t &s : slots)
|
||||
if(s.clientId.load() && (s.feeds.load() & bit)) return true;
|
||||
return false;
|
||||
}
|
||||
ProtoWriter ProtoSocket::beginFeed(proto_type_t type) { return ProtoWriter(beginEvent(type)); }
|
||||
void ProtoSocket::endFeed(proto_feed_t feed) {
|
||||
endFrame(fmt);
|
||||
if(fmt.overflowed() || !this->onLoopTask()) return;
|
||||
if(!finishEvent()) return;
|
||||
const uint8_t bit = 1 << static_cast<uint8_t>(feed);
|
||||
for(proto_slot_t &s : slots) {
|
||||
const uint32_t id = s.asyncId.load();
|
||||
if(id && (s.feeds.load() & bit)) ws.text(id, socketFrameBuffer(), fmt.length());
|
||||
const uint32_t id = s.clientId.load();
|
||||
if(id && (s.feeds.load() & bit)) ws.text(id, socketFrameBuffer(), eventJson.length());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue