Commander1024/ESP32_Serial_Swiss_Army_Knife
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add broker management and writer transfer UI

Browse Source
This commit is contained in:
Commander1024
2026-09-13 13:59:28 +02:00
parent 29a4953df0
commit fa12440606
26 changed files with 1037 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/session_broker.c
+51 -5
View File
@@ -47,6 +47,8 @@ static session_broker_slot_t s_slots[SESSION_BROKER_MAX_CLIENTS];
static session_broker_client_id_t s_writer_id;
static uint32_t s_connected_clients;
static uint64_t s_event_sequence;
/* Saturation disables management confirmations, never ordinary recovery. */
static uint32_t s_writer_generation = 1U;
static session_broker_global_counters_t s_counters;
static bool s_initialized;
@@ -102,6 +104,10 @@ static void broadcast_event_locked(session_broker_event_type_t type,
session_broker_client_id_t client_id,
session_broker_client_id_t writer_id)
{
if ((type == SESSION_BROKER_EVENT_WRITER_GRANTED ||
type == SESSION_BROKER_EVENT_WRITER_RELEASED ||
type == SESSION_BROKER_EVENT_WRITER_REVOKED) && s_writer_generation != UINT32_MAX)
++s_writer_generation;
session_broker_event_t event = {
.sequence = ++s_event_sequence,
.type = type,
@@ -299,7 +305,8 @@ esp_err_t session_broker_connect(session_broker_client_type_t type,
session_broker_slot_t *slot = NULL;
size_t slot_index = 0U;
for (; slot_index < SESSION_BROKER_MAX_CLIENTS; ++slot_index) {
if (!s_slots[slot_index].connected) {
if (!s_slots[slot_index].connected &&
s_slots[slot_index].generation < SESSION_BROKER_MAX_GENERATION) {
slot = &s_slots[slot_index];
break;
}
@@ -319,9 +326,7 @@ esp_err_t session_broker_connect(session_broker_client_type_t type,
}
uint32_t generation = slot->generation + 1U;
if (generation == 0U || generation > SESSION_BROKER_MAX_GENERATION) {
generation = 1U;
}
/* Exhausted slots are retired until reboot: no 29-bit ID reuse. */
xStreamBufferReset(slot->output);
xQueueReset(slot->events);
@@ -470,13 +475,19 @@ esp_err_t session_broker_release_writer(session_broker_client_id_t client_id)
return ESP_OK;
}
esp_err_t session_broker_force_writer(session_broker_client_id_t client_id)
static esp_err_t broker_force_writer(session_broker_client_id_t client_id,
uint32_t expected_generation)
{
if (!s_initialized) {
return ESP_ERR_INVALID_STATE;
}
xSemaphoreTake(s_mutex, portMAX_DELAY);
if (expected_generation && (expected_generation == UINT32_MAX ||
expected_generation != s_writer_generation)) {
xSemaphoreGive(s_mutex);
return ESP_ERR_INVALID_STATE;
}
session_broker_slot_t *new_writer = NULL;
if (client_id != SESSION_BROKER_NO_CLIENT) {
new_writer = find_slot_locked(client_id);
@@ -522,6 +533,41 @@ esp_err_t session_broker_force_writer(session_broker_client_id_t client_id)
return ESP_OK;
}
esp_err_t session_broker_force_writer(session_broker_client_id_t client_id)
{
return broker_force_writer(client_id, 0);
}
esp_err_t session_broker_assign_writer_current(session_broker_client_id_t client_id,
uint32_t generation)
{
if (!client_id || !generation) return ESP_ERR_INVALID_ARG;
return broker_force_writer(client_id, generation);
}
esp_err_t session_broker_get_management_snapshot(session_broker_management_snapshot_t *snapshot)
{
if (!snapshot) return ESP_ERR_INVALID_ARG;
if (!s_initialized) return ESP_ERR_INVALID_STATE;
if (xSemaphoreTake(s_mutex, 0) != pdTRUE) return ESP_ERR_TIMEOUT;
memset(snapshot, 0, sizeof(*snapshot));
snapshot->generation = s_writer_generation;
snapshot->writer_id = s_writer_id;
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
const session_broker_slot_t *slot = &s_slots[i];
if (!slot->connected) continue;
session_broker_management_client_t *client = &snapshot->clients[snapshot->count++];
client->id = slot->id;
client->type = slot->type;
memcpy(client->name, slot->name, sizeof(client->name));
client->pending = xStreamBufferBytesAvailable(slot->output);
client->high_water = slot->counters.output_high_water_bytes;
client->dropped = slot->counters.output_dropped_bytes;
}
xSemaphoreGive(s_mutex);
return ESP_OK;
}
esp_err_t session_broker_force_release_writer(
session_broker_client_id_t expected_writer_id)
{