#include "multimodal/transport_ws.h" #include #include #include #include #include "freertos/FreeRTOS.h" #include "freertos/idf_additions.h" #include "freertos/queue.h" #include "freertos/semphr.h" #include "freertos/task.h" #include "sdkconfig.h" #include "esp_crt_bundle.h" #include "esp_event.h" #include "esp_heap_caps.h" #include "esp_log.h" #include "esp_websocket_client.h" #include "soc/soc_caps.h" #ifndef CONFIG_MULTIMODAL_WS_URI #define CONFIG_MULTIMODAL_WS_URI "wss://dashscope.aliyuncs.com/api-ws/v1/inference" #endif #ifndef CONFIG_MULTIMODAL_CONNECT_TIMEOUT_MS #define CONFIG_MULTIMODAL_CONNECT_TIMEOUT_MS 10000 #endif #ifndef CONFIG_MULTIMODAL_RECONNECT_DELAY_MS #define CONFIG_MULTIMODAL_RECONNECT_DELAY_MS 2000 #endif #ifndef CONFIG_MULTIMODAL_WEBSOCKET_TASK_STACK_BYTES #define CONFIG_MULTIMODAL_WEBSOCKET_TASK_STACK_BYTES 32768 #endif #ifndef CONFIG_MULTIMODAL_WEBSOCKET_TASK_PRIORITY #define CONFIG_MULTIMODAL_WEBSOCKET_TASK_PRIORITY 8 #endif #ifndef CONFIG_MULTIMODAL_WS_DISPATCH_TASK_STACK_BYTES #define CONFIG_MULTIMODAL_WS_DISPATCH_TASK_STACK_BYTES 4096 #endif #ifndef CONFIG_MULTIMODAL_WS_JSON_TASK_STACK_BYTES #define CONFIG_MULTIMODAL_WS_JSON_TASK_STACK_BYTES 8192 #endif #ifndef CONFIG_MULTIMODAL_WS_OPUS_TASK_STACK_BYTES #define CONFIG_MULTIMODAL_WS_OPUS_TASK_STACK_BYTES 12288 #endif #ifndef CONFIG_MULTIMODAL_WS_MSG_QUEUE_LEN #define CONFIG_MULTIMODAL_WS_MSG_QUEUE_LEN 512 #endif #ifndef CONFIG_MULTIMODAL_WS_BUFFER_SIZE #define CONFIG_MULTIMODAL_WS_BUFFER_SIZE 2048 #endif #ifndef CONFIG_MULTIMODAL_WS_PING_INTERVAL_SEC #define CONFIG_MULTIMODAL_WS_PING_INTERVAL_SEC 20 #endif #ifndef CONFIG_MULTIMODAL_WS_PINGPONG_TIMEOUT_SEC #define CONFIG_MULTIMODAL_WS_PINGPONG_TIMEOUT_SEC 120 #endif #ifndef CONFIG_MULTIMODAL_WS_ENABLE_TCP_KEEPALIVE #define CONFIG_MULTIMODAL_WS_ENABLE_TCP_KEEPALIVE 1 #endif #ifndef CONFIG_MULTIMODAL_WS_TCP_KEEPALIVE_IDLE_SEC #define CONFIG_MULTIMODAL_WS_TCP_KEEPALIVE_IDLE_SEC 30 #endif #ifndef CONFIG_MULTIMODAL_WS_TCP_KEEPALIVE_INTERVAL_SEC #define CONFIG_MULTIMODAL_WS_TCP_KEEPALIVE_INTERVAL_SEC 10 #endif #ifndef CONFIG_MULTIMODAL_WS_TCP_KEEPALIVE_COUNT #define CONFIG_MULTIMODAL_WS_TCP_KEEPALIVE_COUNT 3 #endif #ifndef CONFIG_MULTIMODAL_WS_REALTIME_LOCK_WAIT_MS #define CONFIG_MULTIMODAL_WS_REALTIME_LOCK_WAIT_MS 80 #endif #ifndef CONFIG_MULTIMODAL_WS_REALTIME_SEND_TIMEOUT_MS #define CONFIG_MULTIMODAL_WS_REALTIME_SEND_TIMEOUT_MS 300 #endif #ifndef CONFIG_MULTIMODAL_AUDIO_TASK_CORE #define CONFIG_MULTIMODAL_AUDIO_TASK_CORE 1 #endif #ifndef CONFIG_TQ_NETWORK_TASK_CORE #define CONFIG_TQ_NETWORK_TASK_CORE 0 #endif #if CONFIG_TQ_NETWORK_TASK_CORE < 0 #define MULTIMODAL_WS_CORE_ID tskNO_AFFINITY #else #define MULTIMODAL_WS_CORE_ID ((BaseType_t)CONFIG_TQ_NETWORK_TASK_CORE) #endif #define WS_OPCODE_TEXT 0x1 #define WS_OPCODE_BINARY 0x2 #define MULTIMODAL_WS_MSG_MAX_BYTES 4096U #define WS_TEXT_REASSEMBLE_MAX_BYTES (MULTIMODAL_WS_MSG_MAX_BYTES * 8U) #define WS_BINARY_REASSEMBLE_MAX_BYTES MULTIMODAL_WS_MSG_MAX_BYTES #define MULTIMODAL_WS_STACK_MIN 8192 #define MULTIMODAL_WS_STACK_MAX 32768 #define MULTIMODAL_WS_DISPATCH_STACK_MIN 4096 #define MULTIMODAL_WS_TEXT_STACK_MIN 8192 #define MULTIMODAL_WS_BINARY_STACK_MIN 12288 #define MULTIMODAL_WS_DISPATCH_STACK_CAP 4096U #define MULTIMODAL_WS_TEXT_STACK_CAP 8192U #define MULTIMODAL_WS_BINARY_STACK_CAP 12288U #define MULTIMODAL_WS_MSG_QUEUE_CAP 512U #define MULTIMODAL_WS_SEND_TIMEOUT_MS 1200 #define MULTIMODAL_WS_SEND_RETRY_MAX 4 #define MULTIMODAL_WS_SEND_RETRY_DELAY_MS 80 #define MULTIMODAL_WS_TEXT_REALTIME_LOCK_WAIT_MS 80 #define MULTIMODAL_WS_TEXT_REALTIME_SEND_TIMEOUT_MS 300 #define MULTIMODAL_WS_BINARY_GAP_LOG_MS 600 #define MULTIMODAL_WS_BINARY_SUMMARY_INTERVAL 200U #define MULTIMODAL_WS_ASYNC_STOP_TASK_STACK_BYTES 4096 #define MULTIMODAL_WS_ASYNC_STOP_START_GATE_WAIT_MS 5000 #define MULTIMODAL_WS_REALTIME_LOCK_WAIT_MS CONFIG_MULTIMODAL_WS_REALTIME_LOCK_WAIT_MS #define MULTIMODAL_WS_REALTIME_SEND_TIMEOUT_MS CONFIG_MULTIMODAL_WS_REALTIME_SEND_TIMEOUT_MS #define MULTIMODAL_WS_WORKER_STOP_WAIT_MS 1200 #define MULTIMODAL_TLS_WANT_WRITE_ERR (-0x6880) #define MULTIMODAL_TLS_WANT_READ_ERR (-0x6900) #define MULTIMODAL_WS_TOKEN_MAX_LEN 256 static const char *TAG = "multimodal_ws"; typedef struct { uint8_t op_code; uint8_t *data; size_t data_len; int payload_len; int payload_offset; } ws_msg_t; static esp_websocket_client_handle_t s_client; static bool s_connected; static bool s_stop_requested; static portMUX_TYPE s_client_state_lock = portMUX_INITIALIZER_UNLOCKED; static char s_bearer_token[MULTIMODAL_WS_TOKEN_MAX_LEN]; static char s_headers[768]; static multimodal_ws_event_callback_t s_event_cb; static void *s_event_cb_ctx; static portMUX_TYPE s_event_cb_lock = portMUX_INITIALIZER_UNLOCKED; static QueueHandle_t s_ws_msg_queue; static bool s_ws_msg_queue_with_caps; static TaskHandle_t s_ws_dispatch_task; static TaskHandle_t s_ws_async_stop_task; static bool s_ws_async_stop_starting; static volatile bool s_ws_worker_stop_requested; static SemaphoreHandle_t s_ws_send_lock; static TickType_t s_ws_binary_event_last_tick; static TickType_t s_ws_binary_worker_last_tick; static char *s_text_reassembly_buf; static size_t s_text_reassembly_capacity; static size_t s_text_reassembly_filled; static int s_text_reassembly_expected; static uint8_t *s_binary_reassembly_buf; static size_t s_binary_reassembly_capacity; static size_t s_binary_reassembly_filled; static int s_binary_reassembly_expected; static bool s_ws_dispatch_stack_warned; static bool s_ws_text_stack_warned; static bool s_ws_binary_stack_warned; static bool s_ws_dispatch_stack_cap_warned; static bool s_ws_text_stack_cap_warned; static bool s_ws_binary_stack_cap_warned; static bool s_ws_msg_queue_cap_warned; static uint32_t s_ws_binary_event_enqueued; static uint32_t s_ws_binary_worker_processed; static void websocket_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data); static void destroy_queue_with_messages(QueueHandle_t *queue, bool *queue_with_caps); static bool destroy_worker_pipeline_until(TickType_t deadline_tick); static void set_event_callback(multimodal_ws_event_callback_t cb, void *user_data); static void snapshot_event_callback(multimodal_ws_event_callback_t *out_cb, void **out_user_data); static esp_websocket_client_handle_t ws_get_client_handle(void); static void ws_set_client_handle(esp_websocket_client_handle_t client); static bool ws_client_state_snapshot(esp_websocket_client_handle_t *out_client, bool *out_connected, bool *out_stop_requested); static void ws_set_connected(bool connected); static void ws_set_stop_requested(bool stop_requested); static bool is_retryable_send_error(int transport_ret, int saved_errno); static uint32_t normalize_worker_stack_bytes(int configured, uint32_t min_bytes, const char *task_name, bool *warned_once); static uint32_t cap_worker_stack_bytes(uint32_t stack_bytes, uint32_t cap_bytes, const char *task_name, bool *warned_once); static UBaseType_t normalize_queue_depth(int configured, UBaseType_t min_depth, UBaseType_t cap_depth, const char *queue_name, bool *warned_once); static BaseType_t create_ws_task_with_caps(TaskFunction_t task, const char *name, uint32_t stack_bytes, void *arg, UBaseType_t priority, TaskHandle_t *out_handle, bool prefer_psram_stack, BaseType_t core_id); static void *psram_first_malloc(size_t size); static void free_ws_msg(ws_msg_t *msg); static void reset_text_reassembly(void); static void reset_binary_reassembly(void); static bool is_ws_fragmented_payload(int payload_len, int payload_offset, size_t data_len); static void log_queue_drop(const char *queue_name, QueueHandle_t queue, uint8_t op_code, size_t data_len); static void emit_event(multimodal_ws_event_type_t type, const uint8_t *data, size_t data_len); static void process_text_fragment(const ws_msg_t *msg); static void process_binary_fragment(const ws_msg_t *msg); static void handle_ws_msg_direct(ws_msg_t *msg); static void ws_dispatch_task(void *arg); static esp_err_t create_worker_pipeline(void); static int clamp_stack_size(int requested); static size_t unique_stack_candidates(int requested, int *out, size_t out_len); static void reorder_stack_candidates_by_heap_fit(int *candidates, size_t count); static bool ws_transport_ready(void); static esp_err_t start_websocket_client_with_fallback(void); static esp_err_t multimodal_ws_stop_sync_until(TickType_t deadline_tick); static TickType_t ws_stop_deadline_from_timeout(int timeout_ms); static int ws_stop_remaining_ms(TickType_t deadline_tick); static bool ws_has_owned_resources(void); static bool ws_async_stop_inflight(TaskHandle_t *out_task); static bool ws_claim_async_stop(void); static void ws_set_async_stop_task(TaskHandle_t task); static void ws_clear_async_stop_task(TaskHandle_t task); static bool ws_wait_async_stop_done_until(TickType_t deadline_tick); static esp_err_t ws_start_async_stop(void); static void ws_async_stop_task(void *arg); static esp_err_t websocket_send_text_internal(const char *label, const char *text); static esp_err_t websocket_send_text_with_policy(const char *label, const char *text, TickType_t lock_wait, TickType_t send_timeout, bool with_retry); static esp_err_t websocket_send_binary_internal(const uint8_t *data, size_t len, TickType_t lock_wait, TickType_t send_timeout, bool with_retry); static void set_event_callback(multimodal_ws_event_callback_t cb, void *user_data) { portENTER_CRITICAL(&s_event_cb_lock); s_event_cb = cb; s_event_cb_ctx = user_data; portEXIT_CRITICAL(&s_event_cb_lock); } static void snapshot_event_callback(multimodal_ws_event_callback_t *out_cb, void **out_user_data) { multimodal_ws_event_callback_t cb = NULL; void *user_data = NULL; portENTER_CRITICAL(&s_event_cb_lock); cb = s_event_cb; user_data = s_event_cb_ctx; portEXIT_CRITICAL(&s_event_cb_lock); if (out_cb != NULL) { *out_cb = cb; } if (out_user_data != NULL) { *out_user_data = user_data; } } static esp_websocket_client_handle_t ws_get_client_handle(void) { esp_websocket_client_handle_t client; portENTER_CRITICAL(&s_client_state_lock); client = s_client; portEXIT_CRITICAL(&s_client_state_lock); return client; } static void ws_set_client_handle(esp_websocket_client_handle_t client) { portENTER_CRITICAL(&s_client_state_lock); s_client = client; portEXIT_CRITICAL(&s_client_state_lock); } static bool ws_client_state_snapshot(esp_websocket_client_handle_t *out_client, bool *out_connected, bool *out_stop_requested) { esp_websocket_client_handle_t client; bool connected; bool stop_requested; portENTER_CRITICAL(&s_client_state_lock); client = s_client; connected = s_connected; stop_requested = s_stop_requested; portEXIT_CRITICAL(&s_client_state_lock); if (out_client != NULL) { *out_client = client; } if (out_connected != NULL) { *out_connected = connected; } if (out_stop_requested != NULL) { *out_stop_requested = stop_requested; } return client != NULL; } static void ws_set_connected(bool connected) { portENTER_CRITICAL(&s_client_state_lock); s_connected = connected; portEXIT_CRITICAL(&s_client_state_lock); } static void ws_set_stop_requested(bool stop_requested) { portENTER_CRITICAL(&s_client_state_lock); s_stop_requested = stop_requested; portEXIT_CRITICAL(&s_client_state_lock); } /* Split implementation by concern while keeping a single translation unit: * - workers: queue/task/stack helpers and worker pipeline logic * - runtime: websocket lifecycle, event handling, send/stop public APIs */ #include "transport_ws/transport_ws_workers_impl.c" #include "transport_ws/transport_ws_runtime_impl.c"