Add Transport-Neutral Serial Session Broker
This commit is contained in:
@@ -0,0 +1,688 @@
|
||||
#include "session_broker.h"
|
||||
|
||||
#include <string.h>
|
||||
|
||||
#include "freertos/FreeRTOS.h"
|
||||
#include "freertos/queue.h"
|
||||
#include "freertos/semphr.h"
|
||||
#include "freertos/stream_buffer.h"
|
||||
#include "freertos/task.h"
|
||||
#include "serial_service.h"
|
||||
|
||||
#define SESSION_BROKER_RX_CHUNK_SIZE 256U
|
||||
#define SESSION_BROKER_TASK_STACK_SIZE 4096U
|
||||
#define SESSION_BROKER_TASK_PRIORITY 9U
|
||||
#define SESSION_BROKER_IDLE_POLL_MS 5U
|
||||
#define SESSION_BROKER_SLOT_BITS 3U
|
||||
#define SESSION_BROKER_SLOT_MASK ((1U << SESSION_BROKER_SLOT_BITS) - 1U)
|
||||
#define SESSION_BROKER_MAX_GENERATION (UINT32_MAX >> SESSION_BROKER_SLOT_BITS)
|
||||
|
||||
_Static_assert(
|
||||
SESSION_BROKER_MAX_CLIENTS == (1U << SESSION_BROKER_SLOT_BITS),
|
||||
"Client slot bit width must exactly represent SESSION_BROKER_MAX_CLIENTS");
|
||||
|
||||
typedef struct {
|
||||
StreamBufferHandle_t output;
|
||||
QueueHandle_t events;
|
||||
session_broker_client_id_t id;
|
||||
uint32_t generation;
|
||||
session_broker_client_type_t type;
|
||||
char name[SESSION_BROKER_CLIENT_NAME_MAX + 1U];
|
||||
bool connected;
|
||||
session_broker_client_counters_t counters;
|
||||
} session_broker_slot_t;
|
||||
|
||||
static SemaphoreHandle_t s_mutex;
|
||||
static TaskHandle_t s_broker_task;
|
||||
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;
|
||||
static session_broker_global_counters_t s_counters;
|
||||
static bool s_initialized;
|
||||
|
||||
static TickType_t milliseconds_to_ticks(uint32_t milliseconds)
|
||||
{
|
||||
TickType_t ticks = pdMS_TO_TICKS(milliseconds);
|
||||
return (milliseconds > 0U && ticks == 0U) ? 1U : ticks;
|
||||
}
|
||||
|
||||
static bool client_type_valid(session_broker_client_type_t type)
|
||||
{
|
||||
return type >= SESSION_BROKER_CLIENT_CONSOLE &&
|
||||
type <= SESSION_BROKER_CLIENT_INTERNAL;
|
||||
}
|
||||
|
||||
static session_broker_client_id_t make_client_id(size_t slot_index, uint32_t generation)
|
||||
{
|
||||
return (session_broker_client_id_t)((generation << SESSION_BROKER_SLOT_BITS) |
|
||||
(uint32_t)slot_index);
|
||||
}
|
||||
|
||||
static session_broker_slot_t *find_slot_locked(session_broker_client_id_t client_id)
|
||||
{
|
||||
if (client_id == SESSION_BROKER_NO_CLIENT) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
size_t slot_index = client_id & SESSION_BROKER_SLOT_MASK;
|
||||
if (slot_index >= SESSION_BROKER_MAX_CLIENTS) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
session_broker_slot_t *slot = &s_slots[slot_index];
|
||||
if (!slot->connected || slot->id != client_id) {
|
||||
return NULL;
|
||||
}
|
||||
return slot;
|
||||
}
|
||||
|
||||
static void fill_client_snapshot_locked(const session_broker_slot_t *slot,
|
||||
session_broker_client_snapshot_t *snapshot)
|
||||
{
|
||||
snapshot->id = slot->id;
|
||||
snapshot->type = slot->type;
|
||||
memcpy(snapshot->name, slot->name, sizeof(snapshot->name));
|
||||
snapshot->is_writer = slot->id == s_writer_id;
|
||||
snapshot->output_bytes_pending = xStreamBufferBytesAvailable(slot->output);
|
||||
snapshot->events_pending = (uint32_t)uxQueueMessagesWaiting(slot->events);
|
||||
snapshot->counters = slot->counters;
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
session_broker_event_t event = {
|
||||
.sequence = ++s_event_sequence,
|
||||
.type = type,
|
||||
.client_id = client_id,
|
||||
.writer_id = writer_id,
|
||||
};
|
||||
++s_counters.events_generated;
|
||||
|
||||
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
|
||||
session_broker_slot_t *slot = &s_slots[i];
|
||||
if (!slot->connected) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (xQueueSend(slot->events, &event, 0) == pdTRUE) {
|
||||
++slot->counters.events_queued;
|
||||
++s_counters.events_queued;
|
||||
} else {
|
||||
++slot->counters.event_drops;
|
||||
++s_counters.event_drops;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void fan_out_rx_locked(const uint8_t *data, size_t size)
|
||||
{
|
||||
s_counters.uart_rx_bytes += size;
|
||||
|
||||
if (s_connected_clients == 0U) {
|
||||
/* UART RX is always drained, even when nobody can observe the bytes. */
|
||||
s_counters.unobserved_rx_bytes += size;
|
||||
return;
|
||||
}
|
||||
|
||||
/*
|
||||
* The broker task is the sole output-stream writer. Every send is
|
||||
* zero-time, so a slow observer only fills its own bounded stream and
|
||||
* accrues drops; it can never stall UART draining or another observer.
|
||||
*/
|
||||
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
|
||||
session_broker_slot_t *slot = &s_slots[i];
|
||||
if (!slot->connected) {
|
||||
continue;
|
||||
}
|
||||
|
||||
slot->counters.uart_rx_bytes += size;
|
||||
size_t queued = xStreamBufferSend(slot->output, data, size, 0);
|
||||
size_t dropped = size - queued;
|
||||
slot->counters.output_queued_bytes += queued;
|
||||
slot->counters.output_dropped_bytes += dropped;
|
||||
s_counters.output_queued_bytes += queued;
|
||||
s_counters.output_dropped_bytes += dropped;
|
||||
}
|
||||
}
|
||||
|
||||
static void broker_task(void *context)
|
||||
{
|
||||
(void)context;
|
||||
uint8_t data[SESSION_BROKER_RX_CHUNK_SIZE];
|
||||
|
||||
for (;;) {
|
||||
/*
|
||||
* Hold the broker mutex across the serial read and fan-out. This gives
|
||||
* every byte batch an exact order relative to connect/disconnect and
|
||||
* prevents data crossing client-generation boundaries.
|
||||
*/
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
size_t received = serial_service_read(data, sizeof(data));
|
||||
if (received > 0U) {
|
||||
fan_out_rx_locked(data, received);
|
||||
}
|
||||
xSemaphoreGive(s_mutex);
|
||||
|
||||
if (received == 0U) {
|
||||
vTaskDelay(milliseconds_to_ticks(SESSION_BROKER_IDLE_POLL_MS));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void cleanup_allocations(void)
|
||||
{
|
||||
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
|
||||
if (s_slots[i].events != NULL) {
|
||||
vQueueDelete(s_slots[i].events);
|
||||
s_slots[i].events = NULL;
|
||||
}
|
||||
if (s_slots[i].output != NULL) {
|
||||
vStreamBufferDelete(s_slots[i].output);
|
||||
s_slots[i].output = NULL;
|
||||
}
|
||||
}
|
||||
if (s_mutex != NULL) {
|
||||
vSemaphoreDelete(s_mutex);
|
||||
s_mutex = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
esp_err_t session_broker_init(void)
|
||||
{
|
||||
if (s_initialized || s_mutex != NULL) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
memset(s_slots, 0, sizeof(s_slots));
|
||||
memset(&s_counters, 0, sizeof(s_counters));
|
||||
s_writer_id = SESSION_BROKER_NO_CLIENT;
|
||||
s_connected_clients = 0U;
|
||||
s_event_sequence = 0U;
|
||||
s_broker_task = NULL;
|
||||
|
||||
s_mutex = xSemaphoreCreateMutex();
|
||||
if (s_mutex == NULL) {
|
||||
return ESP_ERR_NO_MEM;
|
||||
}
|
||||
|
||||
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
|
||||
s_slots[i].output = xStreamBufferCreate(SESSION_BROKER_OUTPUT_SIZE, 1U);
|
||||
s_slots[i].events = xQueueCreate(SESSION_BROKER_EVENT_QUEUE_LENGTH,
|
||||
sizeof(session_broker_event_t));
|
||||
if (s_slots[i].output == NULL || s_slots[i].events == NULL) {
|
||||
cleanup_allocations();
|
||||
return ESP_ERR_NO_MEM;
|
||||
}
|
||||
}
|
||||
|
||||
if (xTaskCreate(broker_task,
|
||||
"session_broker",
|
||||
SESSION_BROKER_TASK_STACK_SIZE,
|
||||
NULL,
|
||||
SESSION_BROKER_TASK_PRIORITY,
|
||||
&s_broker_task) != pdPASS) {
|
||||
s_broker_task = NULL;
|
||||
cleanup_allocations();
|
||||
return ESP_ERR_NO_MEM;
|
||||
}
|
||||
|
||||
s_initialized = true;
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_connect(session_broker_client_type_t type,
|
||||
const char *name,
|
||||
session_broker_client_id_t *client_id)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
if (!client_type_valid(type) || name == NULL || client_id == NULL) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
size_t name_length = strnlen(name, SESSION_BROKER_CLIENT_NAME_MAX + 1U);
|
||||
if (name_length > SESSION_BROKER_CLIENT_NAME_MAX) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
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) {
|
||||
slot = &s_slots[slot_index];
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NO_MEM;
|
||||
}
|
||||
|
||||
/* DTR on-connect must succeed before this connection becomes observable. */
|
||||
if (s_connected_clients == 0U) {
|
||||
esp_err_t result = serial_service_set_session_active(true);
|
||||
if (result != ESP_OK) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
uint32_t generation = slot->generation + 1U;
|
||||
if (generation == 0U || generation > SESSION_BROKER_MAX_GENERATION) {
|
||||
generation = 1U;
|
||||
}
|
||||
|
||||
xStreamBufferReset(slot->output);
|
||||
xQueueReset(slot->events);
|
||||
memset(&slot->counters, 0, sizeof(slot->counters));
|
||||
slot->generation = generation;
|
||||
slot->id = make_client_id(slot_index, generation);
|
||||
slot->type = type;
|
||||
memset(slot->name, 0, sizeof(slot->name));
|
||||
memcpy(slot->name, name, name_length);
|
||||
slot->connected = true;
|
||||
|
||||
++s_connected_clients;
|
||||
++slot->counters.connections;
|
||||
++s_counters.connections;
|
||||
*client_id = slot->id;
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_CONNECT, slot->id, s_writer_id);
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_disconnect(session_broker_client_id_t client_id)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
|
||||
/* A failed last-client DTR transition leaves the session fully intact. */
|
||||
if (s_connected_clients == 1U) {
|
||||
esp_err_t dtr_error = serial_service_set_session_active(false);
|
||||
if (dtr_error != ESP_OK) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return dtr_error;
|
||||
}
|
||||
}
|
||||
|
||||
bool released_writer = s_writer_id == client_id;
|
||||
slot->connected = false;
|
||||
--s_connected_clients;
|
||||
|
||||
if (released_writer) {
|
||||
s_writer_id = SESSION_BROKER_NO_CLIENT;
|
||||
++slot->counters.writer_releases;
|
||||
++slot->counters.writer_changes;
|
||||
++s_counters.writer_releases;
|
||||
++s_counters.writer_changes;
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_RELEASED,
|
||||
client_id,
|
||||
s_writer_id);
|
||||
}
|
||||
|
||||
++s_counters.disconnections;
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_DISCONNECT, client_id, s_writer_id);
|
||||
|
||||
/* Account for accepted output/events that the disconnect discards. */
|
||||
size_t unread_output = xStreamBufferBytesAvailable(slot->output);
|
||||
UBaseType_t unpopped_events = uxQueueMessagesWaiting(slot->events);
|
||||
slot->counters.output_dropped_bytes += unread_output;
|
||||
slot->counters.event_drops += unpopped_events;
|
||||
s_counters.output_dropped_bytes += unread_output;
|
||||
s_counters.event_drops += unpopped_events;
|
||||
|
||||
/* Zero-time resets are mutex-protected against the broker writer/reader. */
|
||||
xStreamBufferReset(slot->output);
|
||||
xQueueReset(slot->events);
|
||||
slot->id = SESSION_BROKER_NO_CLIENT;
|
||||
slot->name[0] = '\0';
|
||||
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_request_writer(session_broker_client_id_t client_id)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
|
||||
++slot->counters.writer_requests;
|
||||
++s_counters.writer_requests;
|
||||
if (s_writer_id == client_id) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
if (s_writer_id != SESSION_BROKER_NO_CLIENT) {
|
||||
++slot->counters.writer_denials;
|
||||
++s_counters.writer_denials;
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_DENIED,
|
||||
client_id,
|
||||
s_writer_id);
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
s_writer_id = client_id;
|
||||
++slot->counters.writer_grants;
|
||||
++slot->counters.writer_changes;
|
||||
++s_counters.writer_grants;
|
||||
++s_counters.writer_changes;
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_GRANTED,
|
||||
client_id,
|
||||
s_writer_id);
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_release_writer(session_broker_client_id_t client_id)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
if (s_writer_id != client_id) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
s_writer_id = SESSION_BROKER_NO_CLIENT;
|
||||
++slot->counters.writer_releases;
|
||||
++slot->counters.writer_changes;
|
||||
++s_counters.writer_releases;
|
||||
++s_counters.writer_changes;
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_RELEASED,
|
||||
client_id,
|
||||
s_writer_id);
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_force_writer(session_broker_client_id_t client_id)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_slot_t *new_writer = NULL;
|
||||
if (client_id != SESSION_BROKER_NO_CLIENT) {
|
||||
new_writer = find_slot_locked(client_id);
|
||||
if (new_writer == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
}
|
||||
if (s_writer_id == client_id) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
session_broker_client_id_t old_writer_id = s_writer_id;
|
||||
session_broker_slot_t *old_writer = find_slot_locked(old_writer_id);
|
||||
|
||||
/* Every emitted event carries the final atomic ownership state. */
|
||||
s_writer_id = client_id;
|
||||
if (old_writer != NULL) {
|
||||
++old_writer->counters.writer_revocations;
|
||||
++old_writer->counters.writer_changes;
|
||||
++s_counters.writer_revocations;
|
||||
}
|
||||
if (new_writer != NULL) {
|
||||
++new_writer->counters.writer_grants;
|
||||
++new_writer->counters.writer_changes;
|
||||
++s_counters.writer_grants;
|
||||
}
|
||||
++s_counters.writer_changes;
|
||||
|
||||
if (old_writer != NULL) {
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_REVOKED,
|
||||
old_writer_id,
|
||||
s_writer_id);
|
||||
}
|
||||
if (new_writer != NULL) {
|
||||
broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_GRANTED,
|
||||
new_writer->id,
|
||||
s_writer_id);
|
||||
}
|
||||
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
session_broker_client_id_t session_broker_get_writer_id(void)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return SESSION_BROKER_NO_CLIENT;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_client_id_t writer_id = s_writer_id;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return writer_id;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_write(session_broker_client_id_t client_id,
|
||||
const uint8_t *data,
|
||||
size_t size,
|
||||
size_t *accepted)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
if (accepted == NULL || (data == NULL && size > 0U)) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
*accepted = 0U;
|
||||
|
||||
if (xSemaphoreTake(s_mutex, 0) != pdTRUE) {
|
||||
return ESP_ERR_TIMEOUT;
|
||||
}
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
s_counters.tx_rejected_bytes += size;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
if (s_writer_id != client_id) {
|
||||
slot->counters.tx_rejected_bytes += size;
|
||||
s_counters.tx_rejected_bytes += size;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
size_t queued = serial_service_write(data, size);
|
||||
size_t rejected = size - queued;
|
||||
slot->counters.tx_accepted_bytes += queued;
|
||||
slot->counters.tx_rejected_bytes += rejected;
|
||||
s_counters.tx_accepted_bytes += queued;
|
||||
s_counters.tx_rejected_bytes += rejected;
|
||||
*accepted = queued;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_read(session_broker_client_id_t client_id,
|
||||
uint8_t *data,
|
||||
size_t size,
|
||||
size_t *received)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
if (received == NULL || (data == NULL && size > 0U)) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
*received = 0U;
|
||||
|
||||
if (xSemaphoreTake(s_mutex, 0) != pdTRUE) {
|
||||
return ESP_ERR_TIMEOUT;
|
||||
}
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
if (size == 0U) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
size_t count = xStreamBufferReceive(slot->output, data, size, 0);
|
||||
slot->counters.output_read_bytes += count;
|
||||
s_counters.output_read_bytes += count;
|
||||
*received = count;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_pop_event(session_broker_client_id_t client_id,
|
||||
session_broker_event_t *event)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
if (event == NULL) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
|
||||
if (xSemaphoreTake(s_mutex, 0) != pdTRUE) {
|
||||
return ESP_ERR_TIMEOUT;
|
||||
}
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
if (xQueueReceive(slot->events, event, 0) != pdTRUE) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_TIMEOUT;
|
||||
}
|
||||
|
||||
++slot->counters.events_popped;
|
||||
++s_counters.events_popped;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_get_client_snapshot(session_broker_client_id_t client_id,
|
||||
session_broker_client_snapshot_t *snapshot)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
if (snapshot == NULL) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
fill_client_snapshot_locked(slot, snapshot);
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_get_global_snapshot(session_broker_global_snapshot_t *snapshot)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
if (snapshot == NULL) {
|
||||
return ESP_ERR_INVALID_ARG;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
snapshot->writer_id = s_writer_id;
|
||||
snapshot->connected_clients = s_connected_clients;
|
||||
snapshot->latest_event_sequence = s_event_sequence;
|
||||
snapshot->counters = s_counters;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
size_t session_broker_list_clients(session_broker_client_snapshot_t *clients,
|
||||
size_t capacity)
|
||||
{
|
||||
if (!s_initialized || (clients == NULL && capacity > 0U)) {
|
||||
return 0U;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
if (clients == NULL) {
|
||||
size_t count = s_connected_clients;
|
||||
xSemaphoreGive(s_mutex);
|
||||
return count;
|
||||
}
|
||||
|
||||
size_t copied = 0U;
|
||||
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS && copied < capacity; ++i) {
|
||||
if (!s_slots[i].connected) {
|
||||
continue;
|
||||
}
|
||||
fill_client_snapshot_locked(&s_slots[i], &clients[copied]);
|
||||
++copied;
|
||||
}
|
||||
xSemaphoreGive(s_mutex);
|
||||
return copied;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_clear_counters(void)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
memset(&s_counters, 0, sizeof(s_counters));
|
||||
for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
|
||||
memset(&s_slots[i].counters, 0, sizeof(s_slots[i].counters));
|
||||
}
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
|
||||
esp_err_t session_broker_clear_client_counters(session_broker_client_id_t client_id)
|
||||
{
|
||||
if (!s_initialized) {
|
||||
return ESP_ERR_INVALID_STATE;
|
||||
}
|
||||
|
||||
xSemaphoreTake(s_mutex, portMAX_DELAY);
|
||||
session_broker_slot_t *slot = find_slot_locked(client_id);
|
||||
if (slot == NULL) {
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_ERR_NOT_FOUND;
|
||||
}
|
||||
memset(&slot->counters, 0, sizeof(slot->counters));
|
||||
xSemaphoreGive(s_mutex);
|
||||
return ESP_OK;
|
||||
}
|
||||
Reference in New Issue
Block a user