From d6ef9d73bbca1137dc62c89b1010c72953330d92 Mon Sep 17 00:00:00 2001 From: Marius Vikhammer Date: Thu, 9 Jan 2020 17:05:10 +0800 Subject: [PATCH 1/2] websocket: backport of websocket client to v3.3 Backports the websocket client and example to ESP-IDF release 3.3. --- .../esp_websocket_client/CMakeLists.txt | 6 + components/esp_websocket_client/component.mk | 3 + .../esp_websocket_client.c | 671 ++++++++++++++++++ .../include/esp_websocket_client.h | 214 ++++++ .../tcp_transport/include/esp_transport_ws.h | 79 +++ components/tcp_transport/transport_ws.c | 189 ++++- docs/Doxyfile | 2 + .../protocols/esp_websocket_client.rst | 70 ++ docs/en/api-reference/protocols/index.rst | 1 + .../protocols/esp_websocket_client.rst | 1 + examples/protocols/websocket/CMakeLists.txt | 7 + examples/protocols/websocket/Makefile | 8 + examples/protocols/websocket/README.md | 62 ++ examples/protocols/websocket/example_test.py | 207 ++++++ .../protocols/websocket/main/CMakeLists.txt | 4 + .../websocket/main/Kconfig.projbuild | 35 + .../protocols/websocket/main/component.mk | 0 .../websocket/main/websocket_example.c | 167 +++++ examples/protocols/websocket/sdkconfig.ci | 3 + 19 files changed, 1706 insertions(+), 23 deletions(-) create mode 100644 components/esp_websocket_client/CMakeLists.txt create mode 100644 components/esp_websocket_client/component.mk create mode 100644 components/esp_websocket_client/esp_websocket_client.c create mode 100644 components/esp_websocket_client/include/esp_websocket_client.h create mode 100644 docs/en/api-reference/protocols/esp_websocket_client.rst create mode 100644 docs/zh_CN/api-reference/protocols/esp_websocket_client.rst create mode 100644 examples/protocols/websocket/CMakeLists.txt create mode 100644 examples/protocols/websocket/Makefile create mode 100644 examples/protocols/websocket/README.md create mode 100644 examples/protocols/websocket/example_test.py create mode 100644 examples/protocols/websocket/main/CMakeLists.txt create mode 100644 examples/protocols/websocket/main/Kconfig.projbuild create mode 100644 examples/protocols/websocket/main/component.mk create mode 100644 examples/protocols/websocket/main/websocket_example.c create mode 100644 examples/protocols/websocket/sdkconfig.ci diff --git a/components/esp_websocket_client/CMakeLists.txt b/components/esp_websocket_client/CMakeLists.txt new file mode 100644 index 000000000..b346e7c58 --- /dev/null +++ b/components/esp_websocket_client/CMakeLists.txt @@ -0,0 +1,6 @@ +set(COMPONENT_SRCS "esp_websocket_client.c") +set(COMPONENT_ADD_INCLUDEDIRS "include") + +set(COMPONENT_REQUIRES lwip esp-tls tcp_transport nghttp) + +register_component() diff --git a/components/esp_websocket_client/component.mk b/components/esp_websocket_client/component.mk new file mode 100644 index 000000000..7fb6cd504 --- /dev/null +++ b/components/esp_websocket_client/component.mk @@ -0,0 +1,3 @@ +COMPONENT_SRCDIRS := . + +COMPONENT_ADD_INCLUDEDIRS := include \ No newline at end of file diff --git a/components/esp_websocket_client/esp_websocket_client.c b/components/esp_websocket_client/esp_websocket_client.c new file mode 100644 index 000000000..e9b412631 --- /dev/null +++ b/components/esp_websocket_client/esp_websocket_client.c @@ -0,0 +1,671 @@ +// Copyright 2015-2018 Espressif Systems (Shanghai) PTE LTD +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at + +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include + +#include "esp_websocket_client.h" +#include "esp_transport.h" +#include "esp_transport_tcp.h" +#include "esp_transport_ssl.h" +#include "esp_transport_ws.h" +/* using uri parser */ +#include "http_parser.h" +#include "freertos/task.h" +#include "freertos/semphr.h" +#include "freertos/queue.h" +#include "freertos/event_groups.h" +#include "esp_log.h" +#include "esp_timer.h" + +static const char *TAG = "WEBSOCKET_CLIENT"; + +#define WEBSOCKET_TCP_DEFAULT_PORT (80) +#define WEBSOCKET_SSL_DEFAULT_PORT (443) +#define WEBSOCKET_BUFFER_SIZE_BYTE (1024) +#define WEBSOCKET_RECONNECT_TIMEOUT_MS (10*1000) +#define WEBSOCKET_TASK_PRIORITY (5) +#define WEBSOCKET_TASK_STACK (4*1024) +#define WEBSOCKET_NETWORK_TIMEOUT_MS (10*1000) +#define WEBSOCKET_PING_TIMEOUT_MS (10*1000) +#define WEBSOCKET_EVENT_QUEUE_SIZE (1) + +#define ESP_WS_CLIENT_MEM_CHECK(TAG, a, action) if (!(a)) { \ + ESP_LOGE(TAG,"%s:%d (%s): %s", __FILE__, __LINE__, __FUNCTION__, "Memory exhausted"); \ + action; \ + } + +const static int STOPPED_BIT = BIT0; + +ESP_EVENT_DEFINE_BASE(WEBSOCKET_EVENTS); + +typedef struct { + int task_stack; + int task_prio; + char *uri; + char *host; + char *path; + char *scheme; + char *username; + char *password; + int port; + bool auto_reconnect; + void *user_context; + int network_timeout_ms; + char *subprotocol; + char *user_agent; + char *headers; +} websocket_config_storage_t; + +typedef enum { + WEBSOCKET_STATE_ERROR = -1, + WEBSOCKET_STATE_UNKNOW = 0, + WEBSOCKET_STATE_INIT, + WEBSOCKET_STATE_CONNECTED, + WEBSOCKET_STATE_WAIT_TIMEOUT, +} websocket_client_state_t; + +struct esp_websocket_client { + esp_event_loop_handle_t event_handle; + esp_transport_list_handle_t transport_list; + esp_transport_handle_t transport; + websocket_config_storage_t *config; + websocket_client_state_t state; + uint64_t keepalive_tick_ms; + uint64_t reconnect_tick_ms; + uint64_t ping_tick_ms; + int wait_timeout_ms; + int auto_reconnect; + bool run; + EventGroupHandle_t status_bits; + xSemaphoreHandle lock; + char *rx_buffer; + char *tx_buffer; + int buffer_size; + ws_transport_opcodes_t last_opcode; +}; + +static uint64_t _tick_get_ms(void) +{ + return esp_timer_get_time()/1000; +} + +static esp_err_t esp_websocket_client_dispatch_event(esp_websocket_client_handle_t client, + esp_websocket_event_id_t event, + const char *data, + int data_len) +{ + esp_err_t err; + esp_websocket_event_data_t event_data; + + event_data.client = client; + event_data.user_context = client->config->user_context; + + event_data.data_ptr = data; + event_data.data_len = data_len; + event_data.op_code = client->last_opcode; + + if ((err = esp_event_post_to(client->event_handle, + WEBSOCKET_EVENTS, event, + &event_data, + sizeof(esp_websocket_event_data_t), + portMAX_DELAY)) != ESP_OK) { + return err; + } + return esp_event_loop_run(client->event_handle, 0); +} + +static esp_err_t esp_websocket_client_abort_connection(esp_websocket_client_handle_t client) +{ + esp_transport_close(client->transport); + client->wait_timeout_ms = WEBSOCKET_RECONNECT_TIMEOUT_MS; + client->reconnect_tick_ms = _tick_get_ms(); + client->state = WEBSOCKET_STATE_WAIT_TIMEOUT; + ESP_LOGI(TAG, "Reconnect after %d ms", client->wait_timeout_ms); + esp_websocket_client_dispatch_event(client, WEBSOCKET_EVENT_DISCONNECTED, NULL, 0); + return ESP_OK; +} + +static esp_err_t esp_websocket_client_set_config(esp_websocket_client_handle_t client, const esp_websocket_client_config_t *config) +{ + websocket_config_storage_t *cfg = client->config; + cfg->task_prio = config->task_prio; + if (cfg->task_prio <= 0) { + cfg->task_prio = WEBSOCKET_TASK_PRIORITY; + } + + cfg->task_stack = config->task_stack; + if (cfg->task_stack == 0) { + cfg->task_stack = WEBSOCKET_TASK_STACK; + } + + if (config->host) { + cfg->host = strdup(config->host); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->host, return ESP_ERR_NO_MEM); + } + + if (config->port) { + cfg->port = config->port; + } + + if (config->username) { + free(cfg->username); + cfg->username = strdup(config->username); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->username, return ESP_ERR_NO_MEM); + } + + if (config->password) { + free(cfg->password); + cfg->password = strdup(config->password); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->password, return ESP_ERR_NO_MEM); + } + + if (config->uri) { + free(cfg->uri); + cfg->uri = strdup(config->uri); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->uri, return ESP_ERR_NO_MEM); + } + if (config->path) { + free(cfg->path); + cfg->path = strdup(config->path); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->path, return ESP_ERR_NO_MEM); + } + if (config->subprotocol) { + free(cfg->subprotocol); + cfg->subprotocol = strdup(config->subprotocol); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->subprotocol, return ESP_ERR_NO_MEM); + } + if (config->user_agent) { + free(cfg->user_agent); + cfg->user_agent = strdup(config->user_agent); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->user_agent, return ESP_ERR_NO_MEM); + } + if (config->headers) { + free(cfg->headers); + cfg->headers = strdup(config->headers); + ESP_WS_CLIENT_MEM_CHECK(TAG, cfg->headers, return ESP_ERR_NO_MEM); + } + + cfg->network_timeout_ms = WEBSOCKET_NETWORK_TIMEOUT_MS; + cfg->user_context = config->user_context; + cfg->auto_reconnect = true; + if (config->disable_auto_reconnect) { + cfg->auto_reconnect = false; + } + + + return ESP_OK; +} + +static esp_err_t esp_websocket_client_destroy_config(esp_websocket_client_handle_t client) +{ + if (client == NULL) { + return ESP_ERR_INVALID_ARG; + } + websocket_config_storage_t *cfg = client->config; + if (client->config == NULL) { + return ESP_ERR_INVALID_ARG; + } + free(cfg->host); + free(cfg->uri); + free(cfg->path); + free(cfg->scheme); + free(cfg->username); + free(cfg->password); + free(cfg->subprotocol); + free(cfg->user_agent); + free(cfg->headers); + memset(cfg, 0, sizeof(websocket_config_storage_t)); + free(client->config); + client->config = NULL; + return ESP_OK; +} + +static void set_websocket_transport_optional_settings(esp_websocket_client_handle_t client, esp_transport_handle_t trans) +{ + if (trans && client->config->path) { + esp_transport_ws_set_path(trans, client->config->path); + } + if (trans && client->config->subprotocol) { + esp_transport_ws_set_subprotocol(trans, client->config->subprotocol); + } + if (trans && client->config->user_agent) { + esp_transport_ws_set_user_agent(trans, client->config->user_agent); + } + if (trans && client->config->headers) { + esp_transport_ws_set_headers(trans, client->config->headers); + } +} + +esp_websocket_client_handle_t esp_websocket_client_init(const esp_websocket_client_config_t *config) +{ + esp_websocket_client_handle_t client = calloc(1, sizeof(struct esp_websocket_client)); + ESP_WS_CLIENT_MEM_CHECK(TAG, client, return NULL); + + esp_event_loop_args_t event_args = { + .queue_size = WEBSOCKET_EVENT_QUEUE_SIZE, + .task_name = NULL // no task will be created + }; + + if (esp_event_loop_create(&event_args, &client->event_handle) != ESP_OK) { + ESP_LOGE(TAG, "Error create event handler for websocket client"); + free(client); + return NULL; + } + + client->lock = xSemaphoreCreateRecursiveMutex(); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->lock, goto _websocket_init_fail); + + client->config = calloc(1, sizeof(websocket_config_storage_t)); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config, goto _websocket_init_fail); + + client->transport_list = esp_transport_list_init(); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->transport_list, goto _websocket_init_fail); + + esp_transport_handle_t tcp = esp_transport_tcp_init(); + ESP_WS_CLIENT_MEM_CHECK(TAG, tcp, goto _websocket_init_fail); + + esp_transport_set_default_port(tcp, WEBSOCKET_TCP_DEFAULT_PORT); + esp_transport_list_add(client->transport_list, tcp, "_tcp"); // need to save to transport list, for cleanup + + + esp_transport_handle_t ws = esp_transport_ws_init(tcp); + ESP_WS_CLIENT_MEM_CHECK(TAG, ws, goto _websocket_init_fail); + + esp_transport_set_default_port(ws, WEBSOCKET_TCP_DEFAULT_PORT); + esp_transport_list_add(client->transport_list, ws, "ws"); + if (config->transport == WEBSOCKET_TRANSPORT_OVER_TCP) { + asprintf(&client->config->scheme, "ws"); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->scheme, goto _websocket_init_fail); + } + + esp_transport_handle_t ssl = esp_transport_ssl_init(); + ESP_WS_CLIENT_MEM_CHECK(TAG, ssl, goto _websocket_init_fail); + + esp_transport_set_default_port(ssl, WEBSOCKET_SSL_DEFAULT_PORT); + if (config->cert_pem) { + esp_transport_ssl_set_cert_data(ssl, config->cert_pem, strlen(config->cert_pem)); + } + esp_transport_list_add(client->transport_list, ssl, "_ssl"); // need to save to transport list, for cleanup + + esp_transport_handle_t wss = esp_transport_ws_init(ssl); + ESP_WS_CLIENT_MEM_CHECK(TAG, wss, goto _websocket_init_fail); + + esp_transport_set_default_port(wss, WEBSOCKET_SSL_DEFAULT_PORT); + + esp_transport_list_add(client->transport_list, wss, "wss"); + if (config->transport == WEBSOCKET_TRANSPORT_OVER_SSL) { + asprintf(&client->config->scheme, "wss"); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->scheme, goto _websocket_init_fail); + } + + if (config->uri) { + if (esp_websocket_client_set_uri(client, config->uri) != ESP_OK) { + ESP_LOGE(TAG, "Invalid uri"); + goto _websocket_init_fail; + } + } + + if (esp_websocket_client_set_config(client, config) != ESP_OK) { + ESP_LOGE(TAG, "Failed to set the configuration"); + goto _websocket_init_fail; + } + + if (client->config->scheme == NULL) { + asprintf(&client->config->scheme, "ws"); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->scheme, goto _websocket_init_fail); + } + + set_websocket_transport_optional_settings(client, esp_transport_list_get_transport(client->transport_list, "ws")); + set_websocket_transport_optional_settings(client, esp_transport_list_get_transport(client->transport_list, "wss")); + + client->keepalive_tick_ms = _tick_get_ms(); + client->reconnect_tick_ms = _tick_get_ms(); + client->ping_tick_ms = _tick_get_ms(); + + int buffer_size = config->buffer_size; + if (buffer_size <= 0) { + buffer_size = WEBSOCKET_BUFFER_SIZE_BYTE; + } + client->rx_buffer = malloc(buffer_size); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->rx_buffer, { + goto _websocket_init_fail; + }); + client->tx_buffer = malloc(buffer_size); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->tx_buffer, { + goto _websocket_init_fail; + }); + client->status_bits = xEventGroupCreate(); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->status_bits, { + goto _websocket_init_fail; + }); + + client->buffer_size = buffer_size; + return client; + +_websocket_init_fail: + esp_websocket_client_destroy(client); + return NULL; +} + +esp_err_t esp_websocket_client_destroy(esp_websocket_client_handle_t client) +{ + if (client == NULL) { + return ESP_ERR_INVALID_ARG; + } + if (client->run) { + esp_websocket_client_stop(client); + } + if (client->event_handle) { + esp_event_loop_delete(client->event_handle); + } + esp_websocket_client_destroy_config(client); + esp_transport_list_destroy(client->transport_list); + vQueueDelete(client->lock); + free(client->tx_buffer); + free(client->rx_buffer); + if (client->status_bits) { + vEventGroupDelete(client->status_bits); + } + free(client); + client = NULL; + return ESP_OK; +} + +esp_err_t esp_websocket_client_set_uri(esp_websocket_client_handle_t client, const char *uri) +{ + if (client == NULL || uri == NULL) { + return ESP_ERR_INVALID_ARG; + } + struct http_parser_url puri; + http_parser_url_init(&puri); + int parser_status = http_parser_parse_url(uri, strlen(uri), 0, &puri); + if (parser_status != 0) { + ESP_LOGE(TAG, "Error parse uri = %s", uri); + return ESP_FAIL; + } + if (puri.field_data[UF_SCHEMA].len) { + free(client->config->scheme); + asprintf(&client->config->scheme, "%.*s", puri.field_data[UF_SCHEMA].len, uri + puri.field_data[UF_SCHEMA].off); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->scheme, return ESP_ERR_NO_MEM); + } + + if (puri.field_data[UF_HOST].len) { + free(client->config->host); + asprintf(&client->config->host, "%.*s", puri.field_data[UF_HOST].len, uri + puri.field_data[UF_HOST].off); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->host, return ESP_ERR_NO_MEM); + } + + + if (puri.field_data[UF_PATH].len || puri.field_data[UF_QUERY].len) { + free(client->config->path); + if (puri.field_data[UF_QUERY].len == 0) { + asprintf(&client->config->path, "%.*s", puri.field_data[UF_PATH].len, uri + puri.field_data[UF_PATH].off); + } else if (puri.field_data[UF_PATH].len == 0) { + asprintf(&client->config->path, "/?%.*s", puri.field_data[UF_QUERY].len, uri + puri.field_data[UF_QUERY].off); + } else { + asprintf(&client->config->path, "%.*s?%.*s", puri.field_data[UF_PATH].len, uri + puri.field_data[UF_PATH].off, + puri.field_data[UF_QUERY].len, uri + puri.field_data[UF_QUERY].off); + } + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->path, return ESP_ERR_NO_MEM); + } + if (puri.field_data[UF_PORT].off) { + client->config->port = strtol((const char*)(uri + puri.field_data[UF_PORT].off), NULL, 10); + } + + if (puri.field_data[UF_USERINFO].len) { + char *user_info = NULL; + asprintf(&user_info, "%.*s", puri.field_data[UF_USERINFO].len, uri + puri.field_data[UF_USERINFO].off); + if (user_info) { + char *pass = strchr(user_info, ':'); + if (pass) { + pass[0] = 0; //terminal username + pass ++; + free(client->config->password); + client->config->password = strdup(pass); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->password, return ESP_ERR_NO_MEM); + } + free(client->config->username); + client->config->username = strdup(user_info); + ESP_WS_CLIENT_MEM_CHECK(TAG, client->config->username, return ESP_ERR_NO_MEM); + free(user_info); + } else { + return ESP_ERR_NO_MEM; + } + } + return ESP_OK; +} + +static void esp_websocket_client_task(void *pv) +{ + const int lock_timeout = portMAX_DELAY; + int rlen; + esp_websocket_client_handle_t client = (esp_websocket_client_handle_t) pv; + client->run = true; + + //get transport by scheme + client->transport = esp_transport_list_get_transport(client->transport_list, client->config->scheme); + + if (client->transport == NULL) { + ESP_LOGE(TAG, "There are no transports valid, stop websocket client"); + client->run = false; + } + //default port + if (client->config->port == 0) { + client->config->port = esp_transport_get_default_port(client->transport); + } + + client->state = WEBSOCKET_STATE_INIT; + xEventGroupClearBits(client->status_bits, STOPPED_BIT); + int read_select = 0; + while (client->run) { + if (xSemaphoreTakeRecursive(client->lock, lock_timeout) != pdPASS) { + ESP_LOGE(TAG, "Failed to lock ws-client tasks, exitting the task..."); + break; + } + switch ((int)client->state) { + case WEBSOCKET_STATE_INIT: + if (client->transport == NULL) { + ESP_LOGE(TAG, "There are no transport"); + client->run = false; + break; + } + if (esp_transport_connect(client->transport, + client->config->host, + client->config->port, + client->config->network_timeout_ms) < 0) { + ESP_LOGE(TAG, "Error transport connect"); + esp_websocket_client_abort_connection(client); + break; + } + ESP_LOGD(TAG, "Transport connected to %s://%s:%d", client->config->scheme, client->config->host, client->config->port); + + client->state = WEBSOCKET_STATE_CONNECTED; + esp_websocket_client_dispatch_event(client, WEBSOCKET_EVENT_CONNECTED, NULL, 0); + + break; + case WEBSOCKET_STATE_CONNECTED: + if (_tick_get_ms() - client->ping_tick_ms > WEBSOCKET_PING_TIMEOUT_MS) { + client->ping_tick_ms = _tick_get_ms(); + ESP_LOGD(TAG, "Sending PING..."); + esp_transport_ws_send_raw(client->transport, WS_TRANSPORT_OPCODES_PING, NULL, 0, client->config->network_timeout_ms); + } + if (read_select == 0) { + ESP_LOGV(TAG, "Read poll timeout: skipping esp_transport_read()..."); + break; + } + client->ping_tick_ms = _tick_get_ms(); + + rlen = esp_transport_read(client->transport, client->rx_buffer, client->buffer_size, client->config->network_timeout_ms); + if (rlen < 0) { + ESP_LOGE(TAG, "Error read data"); + esp_websocket_client_abort_connection(client); + break; + } + if (rlen >= 0) { + client->last_opcode = esp_transport_ws_get_read_opcode(client->transport); + esp_websocket_client_dispatch_event(client, WEBSOCKET_EVENT_DATA, client->rx_buffer, rlen); + // if a PING message received -> send out the PONG + if (client->last_opcode == WS_TRANSPORT_OPCODES_PING) { + const char *data = (rlen == 0) ? NULL : client->rx_buffer; + esp_transport_ws_send_raw(client->transport, WS_TRANSPORT_OPCODES_PONG, data, rlen, + client->config->network_timeout_ms); + } + } + break; + case WEBSOCKET_STATE_WAIT_TIMEOUT: + + if (!client->config->auto_reconnect) { + client->run = false; + break; + } + if (_tick_get_ms() - client->reconnect_tick_ms > client->wait_timeout_ms) { + client->state = WEBSOCKET_STATE_INIT; + client->reconnect_tick_ms = _tick_get_ms(); + ESP_LOGD(TAG, "Reconnecting..."); + } + break; + } + xSemaphoreGiveRecursive(client->lock); + if (WEBSOCKET_STATE_CONNECTED == client->state) { + read_select = esp_transport_poll_read(client->transport, 1000); //Poll every 1000ms + if (read_select < 0) { + ESP_LOGE(TAG, "Network error: esp_transport_poll_read() returned %d, errno=%d", read_select, errno); + esp_websocket_client_abort_connection(client); + } + } else if (WEBSOCKET_STATE_WAIT_TIMEOUT == client->state) { + // waiting for reconnecting... + vTaskDelay(client->wait_timeout_ms / 2 / portTICK_RATE_MS); + } + } + + esp_transport_close(client->transport); + xEventGroupSetBits(client->status_bits, STOPPED_BIT); + client->state = WEBSOCKET_STATE_UNKNOW; + vTaskDelete(NULL); +} + +esp_err_t esp_websocket_client_start(esp_websocket_client_handle_t client) +{ + if (client == NULL) { + return ESP_ERR_INVALID_ARG; + } + if (client->state >= WEBSOCKET_STATE_INIT) { + ESP_LOGE(TAG, "The client has started"); + return ESP_FAIL; + } + if (xTaskCreate(esp_websocket_client_task, "websocket_task", client->config->task_stack, client, client->config->task_prio, NULL) != pdTRUE) { + ESP_LOGE(TAG, "Error create websocket task"); + return ESP_FAIL; + } + xEventGroupClearBits(client->status_bits, STOPPED_BIT); + return ESP_OK; +} + +esp_err_t esp_websocket_client_stop(esp_websocket_client_handle_t client) +{ + if (client == NULL) { + return ESP_ERR_INVALID_ARG; + } + if (!client->run) { + ESP_LOGW(TAG, "Client was not started"); + return ESP_FAIL; + } + client->run = false; + xEventGroupWaitBits(client->status_bits, STOPPED_BIT, false, true, portMAX_DELAY); + client->state = WEBSOCKET_STATE_UNKNOW; + return ESP_OK; +} + +static int esp_websocket_client_send_with_opcode(esp_websocket_client_handle_t client, ws_transport_opcodes_t opcode, const char *data, int len, TickType_t timeout); + +int esp_websocket_client_send_text(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout) +{ + return esp_websocket_client_send_with_opcode(client, WS_TRANSPORT_OPCODES_TEXT, data, len, timeout); +} + +int esp_websocket_client_send(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout) +{ + return esp_websocket_client_send_with_opcode(client, WS_TRANSPORT_OPCODES_BINARY, data, len, timeout); +} + +int esp_websocket_client_send_bin(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout) +{ + return esp_websocket_client_send_with_opcode(client, WS_TRANSPORT_OPCODES_BINARY, data, len, timeout); +} + +static int esp_websocket_client_send_with_opcode(esp_websocket_client_handle_t client, ws_transport_opcodes_t opcode, const char *data, int len, TickType_t timeout) +{ + int need_write = len; + int wlen = 0, widx = 0; + int ret = ESP_FAIL; + + if (client == NULL || data == NULL || len <= 0) { + ESP_LOGE(TAG, "Invalid arguments"); + return ESP_FAIL; + } + + if (xSemaphoreTakeRecursive(client->lock, timeout) != pdPASS) { + ESP_LOGE(TAG, "Could not lock ws-client within %d timeout", timeout); + return ESP_FAIL; + } + + if (!esp_websocket_client_is_connected(client)) { + ESP_LOGE(TAG, "Websocket client is not connected"); + goto unlock_and_return; + } + + if (client->transport == NULL) { + ESP_LOGE(TAG, "Invalid transport"); + goto unlock_and_return; + } + + while (widx < len) { + if (need_write > client->buffer_size) { + need_write = client->buffer_size; + } + memcpy(client->tx_buffer, data + widx, need_write); + // send with ws specific way and specific opcode + wlen = esp_transport_ws_send_raw(client->transport, opcode, (char *)client->tx_buffer, need_write, + (timeout==portMAX_DELAY)? -1 : timeout * portTICK_PERIOD_MS); + if (wlen <= 0) { + ret = wlen; + ESP_LOGE(TAG, "Network error: esp_transport_write() returned %d, errno=%d", ret, errno); + goto unlock_and_return; + } + widx += wlen; + need_write = len - widx; + } + ret = widx; +unlock_and_return: + xSemaphoreGiveRecursive(client->lock); + return ret; +} + +bool esp_websocket_client_is_connected(esp_websocket_client_handle_t client) +{ + if (client == NULL) { + return false; + } + return client->state == WEBSOCKET_STATE_CONNECTED; +} + +esp_err_t esp_websocket_register_events(esp_websocket_client_handle_t client, + esp_websocket_event_id_t event, + esp_event_handler_t event_handler, + void* event_handler_arg) { + if (client == NULL) { + return ESP_ERR_INVALID_ARG; + } + return esp_event_handler_register_with(client->event_handle, WEBSOCKET_EVENTS, event, event_handler, event_handler_arg); +} diff --git a/components/esp_websocket_client/include/esp_websocket_client.h b/components/esp_websocket_client/include/esp_websocket_client.h new file mode 100644 index 000000000..0f8c64e31 --- /dev/null +++ b/components/esp_websocket_client/include/esp_websocket_client.h @@ -0,0 +1,214 @@ +// Copyright 2015-2018 Espressif Systems (Shanghai) PTE LTD +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at + +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef _ESP_WEBSOCKET_CLIENT_H_ +#define _ESP_WEBSOCKET_CLIENT_H_ + + +#include +#include +#include +#include "freertos/FreeRTOS.h" +#include "esp_err.h" +#include "esp_event.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct esp_websocket_client* esp_websocket_client_handle_t; + +ESP_EVENT_DECLARE_BASE(WEBSOCKET_EVENTS); // declaration of the task events family + +/** + * @brief Websocket Client events id + */ +typedef enum { + WEBSOCKET_EVENT_ANY = -1, + WEBSOCKET_EVENT_ERROR = 0, /*!< This event occurs when there are any errors during execution */ + WEBSOCKET_EVENT_CONNECTED, /*!< Once the Websocket has been connected to the server, no data exchange has been performed */ + WEBSOCKET_EVENT_DISCONNECTED, /*!< The connection has been disconnected */ + WEBSOCKET_EVENT_DATA, /*!< When receiving data from the server, possibly multiple portions of the packet */ + WEBSOCKET_EVENT_MAX +} esp_websocket_event_id_t; + +/** + * @brief Websocket event data + */ +typedef struct { + const char *data_ptr; /*!< Data pointer */ + int data_len; /*!< Data length */ + uint8_t op_code; /*!< Received opcode */ + esp_websocket_client_handle_t client; /*!< esp_websocket_client_handle_t context */ + void *user_context; /*!< user_data context, from esp_websocket_client_config_t user_data */ +} esp_websocket_event_data_t; + +/** + * @brief Websocket Client transport + */ +typedef enum { + WEBSOCKET_TRANSPORT_UNKNOWN = 0x0, /*!< Transport unknown */ + WEBSOCKET_TRANSPORT_OVER_TCP, /*!< Transport over tcp */ + WEBSOCKET_TRANSPORT_OVER_SSL, /*!< Transport over ssl */ +} esp_websocket_transport_t; + +/** + * @brief Websocket client setup configuration + */ +typedef struct { + const char *uri; /*!< Websocket URI, the information on the URI can be overrides the other fields below, if any */ + const char *host; /*!< Domain or IP as string */ + int port; /*!< Port to connect, default depend on esp_websocket_transport_t (80 or 443) */ + const char *username; /*!< Using for Http authentication - Not supported for now */ + const char *password; /*!< Using for Http authentication - Not supported for now */ + const char *path; /*!< HTTP Path, if not set, default is `/` */ + bool disable_auto_reconnect; /*!< Disable the automatic reconnect function when disconnected */ + void *user_context; /*!< HTTP user data context */ + int task_prio; /*!< Websocket task priority */ + int task_stack; /*!< Websocket task stack */ + int buffer_size; /*!< Websocket buffer size */ + const char *cert_pem; /*!< SSL Certification, PEM format as string, if the client requires to verify server */ + esp_websocket_transport_t transport; /*!< Websocket transport type, see `esp_websocket_transport_t */ + char *subprotocol; /*!< Websocket subprotocol */ + char *user_agent; /*!< Websocket user-agent */ + char *headers; /*!< Websocket additional headers */ +} esp_websocket_client_config_t; + +/** + * @brief Start a Websocket session + * This function must be the first function to call, + * and it returns a esp_websocket_client_handle_t that you must use as input to other functions in the interface. + * This call MUST have a corresponding call to esp_websocket_client_destroy when the operation is complete. + * + * @param[in] config The configuration + * + * @return + * - `esp_websocket_client_handle_t` + * - NULL if any errors + */ +esp_websocket_client_handle_t esp_websocket_client_init(const esp_websocket_client_config_t *config); + +/** + * @brief Set URL for client, when performing this behavior, the options in the URL will replace the old ones + * Must stop the WebSocket client before set URI if the client has been connected + * + * @param[in] client The client + * @param[in] uri The uri + * + * @return esp_err_t + */ +esp_err_t esp_websocket_client_set_uri(esp_websocket_client_handle_t client, const char *uri); + +/** + * @brief Open the WebSocket connection + * + * @param[in] client The client + * + * @return esp_err_t + */ +esp_err_t esp_websocket_client_start(esp_websocket_client_handle_t client); + +/** + * @brief Close the WebSocket connection + * + * @param[in] client The client + * + * @return esp_err_t + */ +esp_err_t esp_websocket_client_stop(esp_websocket_client_handle_t client); + +/** + * @brief Destroy the WebSocket connection and free all resources. + * This function must be the last function to call for an session. + * It is the opposite of the esp_websocket_client_init function and must be called with the same handle as input that a esp_websocket_client_init call returned. + * This might close all connections this handle has used. + * + * @param[in] client The client + * + * @return esp_err_t + */ +esp_err_t esp_websocket_client_destroy(esp_websocket_client_handle_t client); + +/** + * @brief Generic write data to the WebSocket connection; defaults to binary send + * + * @param[in] client The client + * @param[in] data The data + * @param[in] len The length + * @param[in] timeout Write data timeout in RTOS ticks + * + * @return + * - Number of data was sent + * - (-1) if any errors + */ +int esp_websocket_client_send(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout); + +/** + * @brief Write binary data to the WebSocket connection (data send with WS OPCODE=02, i.e. binary) + * + * @param[in] client The client + * @param[in] data The data + * @param[in] len The length + * @param[in] timeout Write data timeout in RTOS ticks + * + * @return + * - Number of data was sent + * - (-1) if any errors + */ +int esp_websocket_client_send_bin(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout); + +/** + * @brief Write textual data to the WebSocket connection (data send with WS OPCODE=01, i.e. text) + * + * @param[in] client The client + * @param[in] data The data + * @param[in] len The length + * @param[in] timeout Write data timeout in RTOS ticks + * + * @return + * - Number of data was sent + * - (-1) if any errors + */ +int esp_websocket_client_send_text(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout); + +/** + * @brief Check the WebSocket connection status + * + * @param[in] client The client handle + * + * @return + * - true + * - false + */ +bool esp_websocket_client_is_connected(esp_websocket_client_handle_t client); + +/** + * @brief Register the Websocket Events + * + * @param client The client handle + * @param event The event id + * @param event_handler The callback function + * @param event_handler_arg User context + * @return esp_err_t + */ +esp_err_t esp_websocket_register_events(esp_websocket_client_handle_t client, + esp_websocket_event_id_t event, + esp_event_handler_t event_handler, + void* event_handler_arg); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/components/tcp_transport/include/esp_transport_ws.h b/components/tcp_transport/include/esp_transport_ws.h index 582c5c7da..7251e92e2 100644 --- a/components/tcp_transport/include/esp_transport_ws.h +++ b/components/tcp_transport/include/esp_transport_ws.h @@ -13,6 +13,13 @@ extern "C" { #endif +typedef enum ws_transport_opcodes { + WS_TRANSPORT_OPCODES_TEXT = 0x01, + WS_TRANSPORT_OPCODES_BINARY = 0x02, + WS_TRANSPORT_OPCODES_CLOSE = 0x08, + WS_TRANSPORT_OPCODES_PING = 0x09, + WS_TRANSPORT_OPCODES_PONG = 0x0a, +} ws_transport_opcodes_t; /** * @brief Create web socket transport @@ -23,8 +30,80 @@ extern "C" { */ esp_transport_handle_t esp_transport_ws_init(esp_transport_handle_t parent_handle); +/** + * @brief Set HTTP path to update protocol to websocket + * + * @param t websocket transport handle + * @param path The HTTP Path + */ void esp_transport_ws_set_path(esp_transport_handle_t t, const char *path); +/** + * @brief Set websocket sub protocol header + * + * @param t websocket transport handle + * @param sub_protocol Sub protocol string + * + * @return + * - ESP_OK on success + * - One of the error codes + */ +esp_err_t esp_transport_ws_set_subprotocol(esp_transport_handle_t t, const char *sub_protocol); + +/** + * @brief Set websocket user-agent header + * + * @param t websocket transport handle + * @param sub_protocol user-agent string + * + * @return + * - ESP_OK on success + * - One of the error codes + */ +esp_err_t esp_transport_ws_set_user_agent(esp_transport_handle_t t, const char *user_agent); + +/** + * @brief Set websocket additional headers + * + * @param t websocket transport handle + * @param sub_protocol additional header strings each terminated with \r\n + * + * @return + * - ESP_OK on success + * - One of the error codes + */ +esp_err_t esp_transport_ws_set_headers(esp_transport_handle_t t, const char *headers); + +/** + * @brief Sends websocket raw message with custom opcode and payload + * + * Note that generic esp_transport_write for ws handle sends + * binary massages by default if size is > 0 and + * ping message if message size is set to 0. + * This API is provided to support explicit messages with arbitrary opcode, + * should it be PING, PONG or TEXT message with arbitrary data. + * + * @param[in] t Websocket transport handle + * @param[in] opcode ws operation code + * @param[in] buffer The buffer + * @param[in] len The length + * @param[in] timeout_ms The timeout milliseconds (-1 indicates block forever) + * + * @return + * - Number of bytes was written + * - (-1) if there are any errors, should check errno + */ +int esp_transport_ws_send_raw(esp_transport_handle_t t, ws_transport_opcodes_t opcode, const char *b, int len, int timeout_ms); + +/** + * @brief Returns websocket op-code for last received data + * + * @param t websocket transport handle + * + * @return + * - Received op-code as enum + */ +ws_transport_opcodes_t esp_transport_ws_get_read_opcode(esp_transport_handle_t t); #ifdef __cplusplus diff --git a/components/tcp_transport/transport_ws.c b/components/tcp_transport/transport_ws.c index ec7486526..b0d0eca17 100644 --- a/components/tcp_transport/transport_ws.c +++ b/components/tcp_transport/transport_ws.c @@ -31,9 +31,18 @@ static const char *TAG = "TRANSPORT_WS"; typedef struct { char *path; char *buffer; + char *sub_protocol; + char *user_agent; + char *headers; + uint8_t read_opcode; esp_transport_handle_t parent; } transport_ws_t; +static inline uint8_t ws_get_bin_opcode(ws_transport_opcodes_t opcode) +{ + return (uint8_t)opcode; +} + static esp_transport_handle_t ws_get_payload_transport_handle(esp_transport_handle_t t) { transport_ws_t *ws = esp_transport_get_context_data(t); @@ -89,6 +98,10 @@ static int ws_connect(esp_transport_handle_t t, const char *host, int port, int // Size of base64 coded string is equal '((input_size * 4) / 3) + (input_size / 96) + 6' including Z-term unsigned char client_key[28] = {0}; + // Default values for backwards compatibility + const char *user_agent_ptr = (ws->user_agent)?(ws->user_agent):"ESP32 Websocket Client"; + const char *sub_protocol_ptr = (ws->sub_protocol)?(ws->sub_protocol):"mqtt"; + size_t outlen = 0; mbedtls_base64_encode(client_key, sizeof(client_key), &outlen, random_key, sizeof(random_key)); int len = snprintf(ws->buffer, DEFAULT_WS_BUFFER, @@ -97,25 +110,49 @@ static int ws_connect(esp_transport_handle_t t, const char *host, int port, int "Host: %s:%d\r\n" "Upgrade: websocket\r\n" "Sec-WebSocket-Version: 13\r\n" - "Sec-WebSocket-Protocol: mqtt\r\n" + "Sec-WebSocket-Protocol: %s\r\n" "Sec-WebSocket-Key: %s\r\n" - "User-Agent: ESP32 Websocket Client\r\n\r\n", + "User-Agent: %s\r\n", ws->path, - host, port, - client_key); + host, port, sub_protocol_ptr, + client_key, user_agent_ptr); if (len <= 0 || len >= DEFAULT_WS_BUFFER) { ESP_LOGE(TAG, "Error in request generation, %d", len); return -1; } + if (ws->headers) { + ESP_LOGD(TAG, "headers: %s", ws->headers); + int r = snprintf(ws->buffer + len, DEFAULT_WS_BUFFER - len, "%s", ws->headers); + len += r; + if (r <= 0 || len >= DEFAULT_WS_BUFFER) { + ESP_LOGE(TAG, "Error in request generation" + "(strncpy of headers returned %d, desired request len: %d, buffer size: %d", r, len, DEFAULT_WS_BUFFER); + return -1; + } + } + int r = snprintf(ws->buffer + len, DEFAULT_WS_BUFFER - len, "\r\n"); + len += r; + if (r <= 0 || len >= DEFAULT_WS_BUFFER) { + ESP_LOGE(TAG, "Error in request generation" + "(snprintf of header terminal returned %d, desired request len: %d, buffer size: %d", r, len, DEFAULT_WS_BUFFER); + return -1; + } ESP_LOGD(TAG, "Write upgrate request\r\n%s", ws->buffer); if (esp_transport_write(ws->parent, ws->buffer, len, timeout_ms) <= 0) { ESP_LOGE(TAG, "Error write Upgrade header %s", ws->buffer); return -1; } - if ((len = esp_transport_read(ws->parent, ws->buffer, DEFAULT_WS_BUFFER, timeout_ms)) <= 0) { - ESP_LOGE(TAG, "Error read response for Upgrade header %s", ws->buffer); - return -1; - } + int header_len = 0; + do { + if ((len = esp_transport_read(ws->parent, ws->buffer + header_len, DEFAULT_WS_BUFFER - header_len, timeout_ms)) <= 0) { + ESP_LOGE(TAG, "Error read response for Upgrade header %s", ws->buffer); + return -1; + } + header_len += len; + ws->buffer[header_len] = '\0'; + ESP_LOGD(TAG, "Read header chunk %d, current header size: %d", len, header_len); + } while (NULL == strstr(ws->buffer, "\r\n\r\n") && header_len < DEFAULT_WS_BUFFER); + char *server_key = get_http_header(ws->buffer, "Sec-WebSocket-Accept:"); if (server_key == NULL) { ESP_LOGE(TAG, "Sec-WebSocket-Accept not found"); @@ -144,40 +181,75 @@ static int ws_connect(esp_transport_handle_t t, const char *host, int port, int return 0; } -static int ws_write(esp_transport_handle_t t, const char *buff, int len, int timeout_ms) +static int _ws_write(esp_transport_handle_t t, int opcode, int mask_flag, const char *b, int len, int timeout_ms) { transport_ws_t *ws = esp_transport_get_context_data(t); + char *buffer = (char *)b; char ws_header[MAX_WEBSOCKET_HEADER_SIZE]; char *mask; int header_len = 0, i; - char *buffer = (char *)buff; + int poll_write; if ((poll_write = esp_transport_poll_write(ws->parent, timeout_ms)) <= 0) { + ESP_LOGE(TAG, "Error transport_poll_write"); return poll_write; } - ws_header[header_len++] = WS_OPCODE_BINARY | WS_FIN; + ws_header[header_len++] = opcode; // NOTE: no support for > 16-bit sized messages if (len > 125) { - ws_header[header_len++] = WS_SIZE16 | WS_MASK; + ws_header[header_len++] = WS_SIZE16 | mask_flag; ws_header[header_len++] = (uint8_t)(len >> 8); ws_header[header_len++] = (uint8_t)(len & 0xFF); } else { - ws_header[header_len++] = (uint8_t)(len | WS_MASK); + ws_header[header_len++] = (uint8_t)(len | mask_flag); } - mask = &ws_header[header_len]; - getrandom(ws_header + header_len, 4, 0); - header_len += 4; - for (i = 0; i < len; ++i) { - buffer[i] = (buffer[i] ^ mask[i % 4]); + if (mask_flag) { + mask = &ws_header[header_len]; + getrandom(ws_header + header_len, 4, 0); + header_len += 4; + + for (i = 0; i < len; ++i) { + buffer[i] = (buffer[i] ^ mask[i % 4]); + } } if (esp_transport_write(ws->parent, ws_header, header_len, timeout_ms) != header_len) { ESP_LOGE(TAG, "Error write header"); return -1; } - return esp_transport_write(ws->parent, buffer, len, timeout_ms); + + if (len == 0) { + return 0; + } + + int ret = esp_transport_write(ws->parent, buffer, len, timeout_ms); + // in case of masked transport we have to revert back to the original data, as ws layer + // does not create its own copy of data to be sent + if (mask_flag) { + mask = &ws_header[header_len-4]; + for (i = 0; i < len; ++i) { + buffer[i] = (buffer[i] ^ mask[i % 4]); + } + } + return ret; +} + +int esp_transport_ws_send_raw(esp_transport_handle_t t, ws_transport_opcodes_t opcode, const char *b, int len, int timeout_ms) +{ + uint8_t op_code = ws_get_bin_opcode(opcode); + if (t == NULL) { + ESP_LOGE(TAG, "Transport must be a valid ws handle"); + return ESP_ERR_INVALID_ARG; + } + ESP_LOGD(TAG, "Sending raw ws message with opcode %d", op_code); + return _ws_write(t, op_code | WS_FIN, WS_MASK, b, len, timeout_ms); +} + +static int ws_write(esp_transport_handle_t t, const char *b, int len, int timeout_ms) +{ + return _ws_write(t, WS_OPCODE_BINARY | WS_FIN, WS_MASK, b, len, timeout_ms); } static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ms) @@ -185,7 +257,7 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ transport_ws_t *ws = esp_transport_get_context_data(t); int payload_len; char ws_header[MAX_WEBSOCKET_HEADER_SIZE]; - char *data_ptr = ws_header, opcode, mask, *mask_key = NULL; + char *data_ptr = ws_header, mask, *mask_key = NULL; int rlen; int poll_read; if ((poll_read = esp_transport_poll_read(ws->parent, timeout_ms)) <= 0) { @@ -198,12 +270,12 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ ESP_LOGE(TAG, "Error read data"); return rlen; } - opcode = (*data_ptr & 0x0F); + ws->read_opcode = (*data_ptr & 0x0F); data_ptr ++; mask = ((*data_ptr >> 7) & 0x01); payload_len = (*data_ptr & 0x7F); data_ptr++; - ESP_LOGD(TAG, "Opcode: %d, mask: %d, len: %d\r\n", opcode, mask, payload_len); + ESP_LOGD(TAG, "Opcode: %d, mask: %d, len: %d\r\n", ws->read_opcode , mask, payload_len); if (payload_len == 126) { // headerLen += 2; if ((rlen = esp_transport_read(ws->parent, data_ptr, header, timeout_ms)) <= 0) { @@ -271,6 +343,9 @@ static esp_err_t ws_destroy(esp_transport_handle_t t) transport_ws_t *ws = esp_transport_get_context_data(t); free(ws->buffer); free(ws->path); + free(ws->sub_protocol); + free(ws->user_agent); + free(ws->headers); free(ws); return 0; } @@ -288,7 +363,10 @@ esp_transport_handle_t esp_transport_ws_init(esp_transport_handle_t parent_handl ws->parent = parent_handle; ws->path = strdup("/"); - ESP_TRANSPORT_MEM_CHECK(TAG, ws->path, return NULL); + ESP_TRANSPORT_MEM_CHECK(TAG, ws->path, { + free(ws); + return NULL; + }); ws->buffer = malloc(DEFAULT_WS_BUFFER); ESP_TRANSPORT_MEM_CHECK(TAG, ws->buffer, { free(ws->path); @@ -304,3 +382,68 @@ esp_transport_handle_t esp_transport_ws_init(esp_transport_handle_t parent_handl return t; } +esp_err_t esp_transport_ws_set_subprotocol(esp_transport_handle_t t, const char *sub_protocol) +{ + if (t == NULL) { + return ESP_ERR_INVALID_ARG; + } + transport_ws_t *ws = esp_transport_get_context_data(t); + if (ws->sub_protocol) { + free(ws->sub_protocol); + } + if (sub_protocol == NULL) { + ws->sub_protocol = NULL; + return ESP_OK; + } + ws->sub_protocol = strdup(sub_protocol); + if (ws->sub_protocol == NULL) { + return ESP_ERR_NO_MEM; + } + return ESP_OK; +} + +esp_err_t esp_transport_ws_set_user_agent(esp_transport_handle_t t, const char *user_agent) +{ + if (t == NULL) { + return ESP_ERR_INVALID_ARG; + } + transport_ws_t *ws = esp_transport_get_context_data(t); + if (ws->user_agent) { + free(ws->user_agent); + } + if (user_agent == NULL) { + ws->user_agent = NULL; + return ESP_OK; + } + ws->user_agent = strdup(user_agent); + if (ws->user_agent == NULL) { + return ESP_ERR_NO_MEM; + } + return ESP_OK; +} + +esp_err_t esp_transport_ws_set_headers(esp_transport_handle_t t, const char *headers) +{ + if (t == NULL) { + return ESP_ERR_INVALID_ARG; + } + transport_ws_t *ws = esp_transport_get_context_data(t); + if (ws->headers) { + free(ws->headers); + } + if (headers == NULL) { + ws->headers = NULL; + return ESP_OK; + } + ws->headers = strdup(headers); + if (ws->headers == NULL) { + return ESP_ERR_NO_MEM; + } + return ESP_OK; +} + +ws_transport_opcodes_t esp_transport_ws_get_read_opcode(esp_transport_handle_t t) +{ + transport_ws_t *ws = esp_transport_get_context_data(t); + return ws->read_opcode; +} \ No newline at end of file diff --git a/docs/Doxyfile b/docs/Doxyfile index 919358b18..7a60222a7 100644 --- a/docs/Doxyfile +++ b/docs/Doxyfile @@ -118,6 +118,8 @@ INPUT = \ ../../components/esp_http_client/include/esp_http_client.h \ ../../components/esp_http_server/include/esp_http_server.h \ ../../components/esp_https_server/include/esp_https_server.h \ + ## Websocket Client + ../../components/esp_websocket_client/include/esp_websocket_client.h \ ## ## Provisioning - API Reference ## diff --git a/docs/en/api-reference/protocols/esp_websocket_client.rst b/docs/en/api-reference/protocols/esp_websocket_client.rst new file mode 100644 index 000000000..cd4db2413 --- /dev/null +++ b/docs/en/api-reference/protocols/esp_websocket_client.rst @@ -0,0 +1,70 @@ +ESP WebSocket Client +==================== + +Overview +-------- +The ESP WebSocket client is an implementation of `WebSocket protocol client `_ for ESP32 + +Features +-------- + * supports WebSocket over TCP, SSL with mbedtls + * Easy to setup with URI + * Multiple instances (Multiple clients in one application) + +Configuration +------------- +URI +^^^ + +- Supports ``ws``, ``wss`` schemes +- WebSocket samples: + + - ``ws://websocket.org``: WebSocket over TCP, default port 80 + - ``wss://websocket.org``: WebSocket over SSL, default port 443 + +- Minimal configurations: + +.. code:: c + + const esp_websocket_client_config_t ws_cfg = { + .uri = "ws://websocket.org", + }; + +- If there are any options related to the URI in + ``esp_websocket_client_config_t``, the option defined by the URI will be + overridden. Sample: + +.. code:: c + + const esp_websocket_client_config_t ws_cfg = { + .uri = "ws://websocket.org:123", + .port = 4567, + }; + //WebSocket client will connect to websocket.org using port 4567 + +SSL +^^^ + +- Get certificate from server, example: ``websocket.org`` + ``openssl s_client -showcerts -connect websocket.org:443 /dev/null|openssl x509 -outform PEM >websocket_org.pem`` +- Configuration: + +.. code:: cpp + + const esp_websocket_client_config_t ws_cfg = { + .uri = "wss://websocket.org", + .cert_pem = (const char *)websocket_org_pem_start, + }; + +For more options on ``esp_websocket_client_config_t``, please refer to API reference below + +Application Example +------------------- +Simple WebSocket example that uses esp_websocket_client to establish a websocket connection and send/receive data with the `websocket.org `_ Server: :example:`protocols/websocket`. + + +API Reference +------------- + +.. include:: /_build/inc/esp_websocket_client.inc + diff --git a/docs/en/api-reference/protocols/index.rst b/docs/en/api-reference/protocols/index.rst index 8c3637d9b..79a28f85d 100644 --- a/docs/en/api-reference/protocols/index.rst +++ b/docs/en/api-reference/protocols/index.rst @@ -12,6 +12,7 @@ Application Protocols ASIO ESP-MQTT Modbus slave + Websocket Client Example code for this API section is provided in :example:`protocols` directory of ESP-IDF examples. diff --git a/docs/zh_CN/api-reference/protocols/esp_websocket_client.rst b/docs/zh_CN/api-reference/protocols/esp_websocket_client.rst new file mode 100644 index 000000000..c31856240 --- /dev/null +++ b/docs/zh_CN/api-reference/protocols/esp_websocket_client.rst @@ -0,0 +1 @@ +.. include:: ../../../en/api-reference/protocols/esp_websocket_client.rst diff --git a/examples/protocols/websocket/CMakeLists.txt b/examples/protocols/websocket/CMakeLists.txt new file mode 100644 index 000000000..2bf5323cb --- /dev/null +++ b/examples/protocols/websocket/CMakeLists.txt @@ -0,0 +1,7 @@ +# The following four lines of boilerplate have to be in your project's CMakeLists +# in this exact order for cmake to work correctly +cmake_minimum_required(VERSION 3.5) + +include($ENV{IDF_PATH}/tools/cmake/project.cmake) + +project(websocket-example) diff --git a/examples/protocols/websocket/Makefile b/examples/protocols/websocket/Makefile new file mode 100644 index 000000000..5f9dd681a --- /dev/null +++ b/examples/protocols/websocket/Makefile @@ -0,0 +1,8 @@ +# +# This is a project Makefile. It is assumed the directory this Makefile resides in is a +# project subdirectory. +# +PROJECT_NAME := websocket-example + +include $(IDF_PATH)/make/project.mk + diff --git a/examples/protocols/websocket/README.md b/examples/protocols/websocket/README.md new file mode 100644 index 000000000..e49bda81a --- /dev/null +++ b/examples/protocols/websocket/README.md @@ -0,0 +1,62 @@ +# Websocket Sample application + +(See the README.md file in the upper level 'examples' directory for more information about examples.) +This example will shows how to set up and communicate over a websocket. + +## How to Use Example + +### Hardware Required + +This example can be executed on any ESP32 board, the only required interface is WiFi and connection to internet or a local server. + +### Configure the project + +``` +make menuconfig +``` + +* Set serial port under Serial Flasher Options. + +* Set ssid and password for the board to connect to AP. + +### Build and Flash + +Build the project and flash it to the board, then run monitor tool to view serial output: + +``` +make -j4 flash monitor +``` + +(To exit the serial monitor, type ``Ctrl-]``.) + +See the Getting Started Guide for full steps to configure and use ESP-IDF to build projects. + +## Example Output + +``` +I (482) system_api: Base MAC address is not set, read default base MAC address from BLK0 of EFUSE +I (2492) example_connect: Ethernet Link Up +I (4472) tcpip_adapter: eth ip: 192.168.2.137, mask: 255.255.255.0, gw: 192.168.2.2 +I (4472) example_connect: Connected to Ethernet +I (4472) example_connect: IPv4 address: 192.168.2.137 +I (4472) example_connect: IPv6 address: fe80:0000:0000:0000:bedd:c2ff:fed4:a92b +I (4482) WEBSOCKET: Connecting to ws://echo.websocket.org... +I (5012) WEBSOCKET: WEBSOCKET_EVENT_CONNECTED +I (5492) WEBSOCKET: Sending hello 0000 +I (6052) WEBSOCKET: WEBSOCKET_EVENT_DATA +W (6052) WEBSOCKET: Received=hello 0000 + +I (6492) WEBSOCKET: Sending hello 0001 +I (7052) WEBSOCKET: WEBSOCKET_EVENT_DATA +W (7052) WEBSOCKET: Received=hello 0001 + +I (7492) WEBSOCKET: Sending hello 0002 +I (8082) WEBSOCKET: WEBSOCKET_EVENT_DATA +W (8082) WEBSOCKET: Received=hello 0002 + +I (8492) WEBSOCKET: Sending hello 0003 +I (9152) WEBSOCKET: WEBSOCKET_EVENT_DATA +W (9162) WEBSOCKET: Received=hello 0003 + +``` + diff --git a/examples/protocols/websocket/example_test.py b/examples/protocols/websocket/example_test.py new file mode 100644 index 000000000..eeb0e6034 --- /dev/null +++ b/examples/protocols/websocket/example_test.py @@ -0,0 +1,207 @@ +from __future__ import print_function +from __future__ import unicode_literals +import re +import os +import socket +import hashlib +import base64 +import sys +from threading import Thread + +try: + import IDF +except Exception: + # this is a test case write with tiny-test-fw. + # to run test cases outside tiny-test-fw, + # we need to set environment variable `TEST_FW_PATH`, + # then get and insert `TEST_FW_PATH` to sys path before import FW module + test_fw_path = os.getenv("TEST_FW_PATH") + if test_fw_path and test_fw_path not in sys.path: + sys.path.insert(0, test_fw_path) + import IDF + +import DUT + + +def get_my_ip(): + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + # doesn't even have to be reachable + s.connect(('10.255.255.255', 1)) + IP = s.getsockname()[0] + except Exception: + IP = '127.0.0.1' + finally: + s.close() + return IP + + +# Simple Websocket server for testing purposes +class Websocket: + HEADER_LEN = 6 + + def __init__(self, port): + self.port = port + self.socket = socket.socket() + self.socket.settimeout(10.0) + + def __enter__(self): + try: + self.socket.bind(('', self.port)) + except socket.error as e: + print("Bind failed:{}".format(e)) + raise + + self.socket.listen(1) + self.server_thread = Thread(target=self.run_server) + self.server_thread.start() + + def __exit__(self, exc_type, exc_value, traceback): + self.server_thread.join() + self.socket.close() + self.conn.close() + + def run_server(self): + self.conn, address = self.socket.accept() # accept new connection + self.conn.settimeout(10.0) + print("Connection from: {}".format(address)) + + self.establish_connection() + + # Echo data until client closes connection + self.echo_data() + + def establish_connection(self): + while True: + try: + # receive data stream. it won't accept data packet greater than 1024 bytes + data = self.conn.recv(1024).decode() + if not data: + # exit if data is not received + raise + + if "Upgrade: websocket" in data and "Connection: Upgrade" in data: + self.handshake(data) + return + except socket.error as err: + print("Unable to establish a websocket connection: {}, {}".format(err)) + raise + + def handshake(self, data): + # Magic string from RFC + MAGIC_STRING = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + headers = data.split("\r\n") + + for header in headers: + if "Sec-WebSocket-Key" in header: + client_key = header.split()[1] + + if client_key: + resp_key = client_key + MAGIC_STRING + resp_key = base64.standard_b64encode(hashlib.sha1(resp_key.encode()).digest()) + + resp = "HTTP/1.1 101 Switching Protocols\r\n" + \ + "Upgrade: websocket\r\n" + \ + "Connection: Upgrade\r\n" + \ + "Sec-WebSocket-Accept: {}\r\n\r\n".format(resp_key.decode()) + + self.conn.send(resp.encode()) + + def echo_data(self): + while(True): + try: + header = bytearray(self.conn.recv(self.HEADER_LEN, socket.MSG_WAITALL)) + if not header: + # exit if data is not received + return + + # Remove mask bit + payload_len = ~(1 << 7) & header[1] + + payload = bytearray(self.conn.recv(payload_len, socket.MSG_WAITALL)) + frame = header + payload + + decoded_payload = self.decode_frame(frame) + + echo_frame = self.encode_frame(decoded_payload) + self.conn.send(echo_frame) + except socket.error as err: + print("Stopped echoing data: {}".format(err)) + + def decode_frame(self, frame): + # Mask out MASK bit from payload length, this len is only valid for short messages (<126) + payload_len = ~(1 << 7) & frame[1] + + mask = frame[2:self.HEADER_LEN] + + encrypted_payload = frame[self.HEADER_LEN:self.HEADER_LEN + payload_len] + payload = bytearray() + + for i in range(payload_len): + payload.append(encrypted_payload[i] ^ mask[i % 4]) + + return payload + + def encode_frame(self, payload): + # Set FIN = 1 and OP_CODE = 1 (text) + header = (1 << 7) | (1 << 0) + + frame = bytearray([header]) + frame.append(len(payload)) + frame += payload + + return frame + + +def test_echo(dut): + dut.expect("WEBSOCKET_EVENT_CONNECTED") + for i in range(0, 10): + dut.expect(re.compile(r"Received=hello (\d)")) + dut.expect("Websocket Stopped") + + +@IDF.idf_example_test(env_tag="Example_WIFI") +def test_examples_protocol_websocket(env, extra_data): + """ + steps: + 1. join AP + 2. connect to uri specified in the config + 3. send and receive data + """ + dut1 = env.get_dut("websocket", "examples/protocols/websocket") + # check and log bin size + binary_file = os.path.join(dut1.app.binary_path, "websocket-example.bin") + bin_size = os.path.getsize(binary_file) + IDF.log_performance("websocket_bin_size", "{}KB".format(bin_size // 1024)) + IDF.check_performance("websocket_bin_size", bin_size // 1024) + + try: + if "CONFIG_WEBSOCKET_URI_FROM_STDIN" in dut1.app.get_sdkconfig(): + uri_from_stdin = True + else: + uri = dut1.app.get_sdkconfig()["CONFIG_WEBSOCKET_URI"].strip('"') + uri_from_stdin = False + + except Exception: + print('ENV_TEST_FAILURE: Cannot find uri settings in sdkconfig') + raise + + # start test + dut1.start_app() + + if uri_from_stdin: + server_port = 4455 + with Websocket(server_port): + uri = "ws://{}:{}".format(get_my_ip(), server_port) + print("DUT connecting to {}".format(uri)) + dut1.expect("Please enter uri of websocket endpoint", timeout=30) + dut1.write(uri) + test_echo(dut1) + + else: + print("DUT connecting to {}".format(uri)) + test_echo(dut1) + + +if __name__ == '__main__': + test_examples_protocol_websocket() diff --git a/examples/protocols/websocket/main/CMakeLists.txt b/examples/protocols/websocket/main/CMakeLists.txt new file mode 100644 index 000000000..caf642155 --- /dev/null +++ b/examples/protocols/websocket/main/CMakeLists.txt @@ -0,0 +1,4 @@ +set(COMPONENT_SRCS "websocket_example.c") +set(COMPONENT_ADD_INCLUDEDIRS ".") + +register_component() diff --git a/examples/protocols/websocket/main/Kconfig.projbuild b/examples/protocols/websocket/main/Kconfig.projbuild new file mode 100644 index 000000000..c6fd92005 --- /dev/null +++ b/examples/protocols/websocket/main/Kconfig.projbuild @@ -0,0 +1,35 @@ +menu "Example Configuration" + + config WIFI_SSID + string "WiFi SSID" + default "myssid" + help + SSID (network name) for the example to connect to. + + config WIFI_PASSWORD + string "WiFi Password" + default "mypassword" + help + WiFi password (WPA or WPA2) for the example to use. + + choice WEBSOCKET_URI_SOURCE + prompt "Websocket URI source" + default WEBSOCKET_URI_FROM_STRING + help + Selects the source of the URI used in the example. + + config WEBSOCKET_URI_FROM_STRING + bool "From string" + + config WEBSOCKET_URI_FROM_STDIN + bool "From stdin" + endchoice + + config WEBSOCKET_URI + string "Websocket endpoint URI" + depends on WEBSOCKET_URI_FROM_STRING + default "ws://echo.websocket.org" + help + URL of websocket endpoint this example connects to and sends echo + +endmenu diff --git a/examples/protocols/websocket/main/component.mk b/examples/protocols/websocket/main/component.mk new file mode 100644 index 000000000..e69de29bb diff --git a/examples/protocols/websocket/main/websocket_example.c b/examples/protocols/websocket/main/websocket_example.c new file mode 100644 index 000000000..d74474914 --- /dev/null +++ b/examples/protocols/websocket/main/websocket_example.c @@ -0,0 +1,167 @@ +/* ESP Websocket Client Example + + This example code is in the Public Domain (or CC0 licensed, at your option.) + + Unless required by applicable law or agreed to in writing, this + software is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR + CONDITIONS OF ANY KIND, either express or implied. +*/ + + +#include +#include "esp_wifi.h" +#include "esp_system.h" +#include "nvs_flash.h" +#include "esp_event_loop.h" + +#include "freertos/FreeRTOS.h" +#include "freertos/task.h" +#include "freertos/event_groups.h" + + +#include "esp_log.h" +#include "esp_websocket_client.h" +#include "esp_event.h" + +static const char *TAG = "WEBSOCKET"; + +static EventGroupHandle_t wifi_event_group; +const static int CONNECTED_BIT = BIT0; + +#if CONFIG_WEBSOCKET_URI_FROM_STDIN +static void get_string(char *line, size_t size) +{ + int count = 0; + while (count < size) { + int c = fgetc(stdin); + if (c == '\n') { + line[count] = '\0'; + break; + } else if (c > 0 && c < 127) { + line[count] = c; + ++count; + } + vTaskDelay(10 / portTICK_PERIOD_MS); + } +} + +#endif /* CONFIG_WEBSOCKET_URI_FROM_STDIN */ + +static void websocket_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) +{ + esp_websocket_event_data_t *data = (esp_websocket_event_data_t *)event_data; + switch (event_id) { + case WEBSOCKET_EVENT_CONNECTED: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_CONNECTED"); + break; + case WEBSOCKET_EVENT_DISCONNECTED: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_DISCONNECTED"); + break; + case WEBSOCKET_EVENT_DATA: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_DATA"); + ESP_LOGI(TAG, "Received opcode=%d", data->op_code); + ESP_LOGW(TAG, "Received=%.*s\r\n", data->data_len, (char*)data->data_ptr); + break; + case WEBSOCKET_EVENT_ERROR: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_ERROR"); + break; + } +} + +static esp_err_t wifi_event_handler(void *ctx, system_event_t *event) +{ + switch (event->event_id) { + case SYSTEM_EVENT_STA_START: + esp_wifi_connect(); + break; + case SYSTEM_EVENT_STA_GOT_IP: + xEventGroupSetBits(wifi_event_group, CONNECTED_BIT); + + break; + case SYSTEM_EVENT_STA_DISCONNECTED: + esp_wifi_connect(); + xEventGroupClearBits(wifi_event_group, CONNECTED_BIT); + break; + default: + break; + } + return ESP_OK; +} + +static void wifi_init(void) +{ + tcpip_adapter_init(); + wifi_event_group = xEventGroupCreate(); + ESP_ERROR_CHECK(esp_event_loop_init(wifi_event_handler, NULL)); + wifi_init_config_t cfg = WIFI_INIT_CONFIG_DEFAULT(); + ESP_ERROR_CHECK(esp_wifi_init(&cfg)); + ESP_ERROR_CHECK(esp_wifi_set_storage(WIFI_STORAGE_RAM)); + wifi_config_t wifi_config = { + .sta = { + .ssid = CONFIG_WIFI_SSID, + .password = CONFIG_WIFI_PASSWORD, + }, + }; + ESP_ERROR_CHECK(esp_wifi_set_mode(WIFI_MODE_STA)); + ESP_ERROR_CHECK(esp_wifi_set_config(ESP_IF_WIFI_STA, &wifi_config)); + ESP_LOGI(TAG, "start the WIFI SSID:[%s]", CONFIG_WIFI_SSID); + ESP_ERROR_CHECK(esp_wifi_start()); + ESP_LOGI(TAG, "Waiting for wifi"); + xEventGroupWaitBits(wifi_event_group, CONNECTED_BIT, false, true, portMAX_DELAY); +} + +static void websocket_app_start(void) +{ + esp_websocket_client_config_t websocket_cfg = {}; + + #if CONFIG_WEBSOCKET_URI_FROM_STDIN + char line[128]; + + ESP_LOGI(TAG, "Please enter uri of websocket endpoint"); + get_string(line, sizeof(line)); + + websocket_cfg.uri = line; + ESP_LOGI(TAG, "Endpoint uri: %s\n", line); + + #else + websocket_cfg.uri = CONFIG_WEBSOCKET_URI; + + #endif /* CONFIG_WEBSOCKET_URI_FROM_STDIN */ + + ESP_LOGI(TAG, "Connecting to %s...", websocket_cfg.uri); + + esp_websocket_client_handle_t client = esp_websocket_client_init(&websocket_cfg); + esp_websocket_register_events(client, WEBSOCKET_EVENT_ANY, websocket_event_handler, (void *)client); + + esp_websocket_client_start(client); + char data[32]; + int i = 0; + while (i < 10) { + if (esp_websocket_client_is_connected(client)) { + int len = sprintf(data, "hello %04d", i++); + ESP_LOGI(TAG, "Sending %s", data); + esp_websocket_client_send(client, data, len, portMAX_DELAY); + } + vTaskDelay(1000 / portTICK_RATE_MS); + } + // Give server some time to respond before closing + vTaskDelay(3000 / portTICK_RATE_MS); + esp_websocket_client_stop(client); + ESP_LOGI(TAG, "Websocket Stopped"); + esp_websocket_client_destroy(client); +} + +void app_main(void) +{ + ESP_LOGI(TAG, "[APP] Startup.."); + ESP_LOGI(TAG, "[APP] Free memory: %d bytes", esp_get_free_heap_size()); + ESP_LOGI(TAG, "[APP] IDF version: %s", esp_get_idf_version()); + + esp_log_level_set("*", ESP_LOG_INFO); + esp_log_level_set("WEBSOCKET_CLIENT", ESP_LOG_DEBUG); + esp_log_level_set("TRANS_TCP", ESP_LOG_DEBUG); + + nvs_flash_init(); + wifi_init(); + websocket_app_start(); +} \ No newline at end of file diff --git a/examples/protocols/websocket/sdkconfig.ci b/examples/protocols/websocket/sdkconfig.ci new file mode 100644 index 000000000..a0b7712a2 --- /dev/null +++ b/examples/protocols/websocket/sdkconfig.ci @@ -0,0 +1,3 @@ +CONFIG_WEBSOCKET_URI_FROM_STDIN=y +CONFIG_WEBSOCKET_URI_FROM_STRING=n + From b56012783c62d125cedfaf6199eb8a862558910f Mon Sep 17 00:00:00 2001 From: Marius Vikhammer Date: Fri, 20 Mar 2020 11:07:07 +0800 Subject: [PATCH 2/2] tcp_transport/ws_client: websockets now correctly handle messages longer than buffer transport_ws can now be read multiple times in a row to read frames larger than the buffer. Added reporting of total payload length and offset to the user in websocket_client. Added local example test for long messages. Closes IDF-1083 --- .../esp_websocket_client.c | 59 +++++--- .../include/esp_websocket_client.h | 16 +- .../tcp_transport/include/esp_transport_ws.h | 11 ++ components/tcp_transport/transport_ws.c | 143 ++++++++++++++---- examples/protocols/websocket/README.md | 6 - examples/protocols/websocket/example_test.py | 117 ++++++++++---- .../websocket/main/websocket_example.c | 62 +++++--- 7 files changed, 305 insertions(+), 109 deletions(-) diff --git a/components/esp_websocket_client/esp_websocket_client.c b/components/esp_websocket_client/esp_websocket_client.c index e9b412631..d00bdbbe9 100644 --- a/components/esp_websocket_client/esp_websocket_client.c +++ b/components/esp_websocket_client/esp_websocket_client.c @@ -93,6 +93,8 @@ struct esp_websocket_client { char *tx_buffer; int buffer_size; ws_transport_opcodes_t last_opcode; + int payload_len; + int payload_offset; }; static uint64_t _tick_get_ms(void) @@ -101,19 +103,20 @@ static uint64_t _tick_get_ms(void) } static esp_err_t esp_websocket_client_dispatch_event(esp_websocket_client_handle_t client, - esp_websocket_event_id_t event, - const char *data, - int data_len) + esp_websocket_event_id_t event, + const char *data, + int data_len) { esp_err_t err; esp_websocket_event_data_t event_data; event_data.client = client; event_data.user_context = client->config->user_context; - event_data.data_ptr = data; event_data.data_len = data_len; event_data.op_code = client->last_opcode; + event_data.payload_len = client->payload_len; + event_data.payload_offset = client->payload_offset; if ((err = esp_event_post_to(client->event_handle, WEBSOCKET_EVENTS, event, @@ -446,10 +449,38 @@ esp_err_t esp_websocket_client_set_uri(esp_websocket_client_handle_t client, con return ESP_OK; } +static esp_err_t esp_websocket_client_recv(esp_websocket_client_handle_t client) +{ + int rlen; + client->payload_offset = 0; + do { + rlen = esp_transport_read(client->transport, client->rx_buffer, client->buffer_size, client->config->network_timeout_ms); + if (rlen < 0) { + ESP_LOGE(TAG, "Error read data"); + esp_websocket_client_abort_connection(client); + return ESP_FAIL; + } + client->payload_len = esp_transport_ws_get_read_payload_len(client->transport); + client->last_opcode = esp_transport_ws_get_read_opcode(client->transport); + + esp_websocket_client_dispatch_event(client, WEBSOCKET_EVENT_DATA, client->rx_buffer, rlen); + + client->payload_offset += rlen; + } while (client->payload_offset < client->payload_len); + + // if a PING message received -> send out the PONG, this will not work for PING messages with payload longer than buffer len + if (client->last_opcode == WS_TRANSPORT_OPCODES_PING) { + const char *data = (client->payload_len == 0) ? NULL : client->rx_buffer; + esp_transport_ws_send_raw(client->transport, WS_TRANSPORT_OPCODES_PONG, data, client->payload_len, + client->config->network_timeout_ms); + } + + return ESP_OK; +} + static void esp_websocket_client_task(void *pv) { const int lock_timeout = portMAX_DELAY; - int rlen; esp_websocket_client_handle_t client = (esp_websocket_client_handle_t) pv; client->run = true; @@ -506,22 +537,11 @@ static void esp_websocket_client_task(void *pv) } client->ping_tick_ms = _tick_get_ms(); - rlen = esp_transport_read(client->transport, client->rx_buffer, client->buffer_size, client->config->network_timeout_ms); - if (rlen < 0) { - ESP_LOGE(TAG, "Error read data"); + if (esp_websocket_client_recv(client) == ESP_FAIL) { + ESP_LOGE(TAG, "Error receive data"); esp_websocket_client_abort_connection(client); break; } - if (rlen >= 0) { - client->last_opcode = esp_transport_ws_get_read_opcode(client->transport); - esp_websocket_client_dispatch_event(client, WEBSOCKET_EVENT_DATA, client->rx_buffer, rlen); - // if a PING message received -> send out the PONG - if (client->last_opcode == WS_TRANSPORT_OPCODES_PING) { - const char *data = (rlen == 0) ? NULL : client->rx_buffer; - esp_transport_ws_send_raw(client->transport, WS_TRANSPORT_OPCODES_PONG, data, rlen, - client->config->network_timeout_ms); - } - } break; case WEBSOCKET_STATE_WAIT_TIMEOUT: @@ -663,7 +683,8 @@ bool esp_websocket_client_is_connected(esp_websocket_client_handle_t client) esp_err_t esp_websocket_register_events(esp_websocket_client_handle_t client, esp_websocket_event_id_t event, esp_event_handler_t event_handler, - void* event_handler_arg) { + void *event_handler_arg) +{ if (client == NULL) { return ESP_ERR_INVALID_ARG; } diff --git a/components/esp_websocket_client/include/esp_websocket_client.h b/components/esp_websocket_client/include/esp_websocket_client.h index 0f8c64e31..ae8cc8a4b 100644 --- a/components/esp_websocket_client/include/esp_websocket_client.h +++ b/components/esp_websocket_client/include/esp_websocket_client.h @@ -27,7 +27,7 @@ extern "C" { #endif -typedef struct esp_websocket_client* esp_websocket_client_handle_t; +typedef struct esp_websocket_client *esp_websocket_client_handle_t; ESP_EVENT_DECLARE_BASE(WEBSOCKET_EVENTS); // declaration of the task events family @@ -47,11 +47,13 @@ typedef enum { * @brief Websocket event data */ typedef struct { - const char *data_ptr; /*!< Data pointer */ - int data_len; /*!< Data length */ - uint8_t op_code; /*!< Received opcode */ - esp_websocket_client_handle_t client; /*!< esp_websocket_client_handle_t context */ - void *user_context; /*!< user_data context, from esp_websocket_client_config_t user_data */ + const char *data_ptr; /*!< Data pointer */ + int data_len; /*!< Data length */ + uint8_t op_code; /*!< Received opcode */ + esp_websocket_client_handle_t client; /*!< esp_websocket_client_handle_t context */ + void *user_context; /*!< user_data context, from esp_websocket_client_config_t user_data */ + int payload_len; /*!< Total payload length, payloads exceeding buffer will be posted through multiple events */ + int payload_offset; /*!< Actual offset for the data associated with this event */ } esp_websocket_event_data_t; /** @@ -205,7 +207,7 @@ bool esp_websocket_client_is_connected(esp_websocket_client_handle_t client); esp_err_t esp_websocket_register_events(esp_websocket_client_handle_t client, esp_websocket_event_id_t event, esp_event_handler_t event_handler, - void* event_handler_arg); + void *event_handler_arg); #ifdef __cplusplus } diff --git a/components/tcp_transport/include/esp_transport_ws.h b/components/tcp_transport/include/esp_transport_ws.h index 7251e92e2..5e5405791 100644 --- a/components/tcp_transport/include/esp_transport_ws.h +++ b/components/tcp_transport/include/esp_transport_ws.h @@ -14,6 +14,7 @@ extern "C" { #endif typedef enum ws_transport_opcodes { + WS_TRANSPORT_OPCODES_CONT = 0x00, WS_TRANSPORT_OPCODES_TEXT = 0x01, WS_TRANSPORT_OPCODES_BINARY = 0x02, WS_TRANSPORT_OPCODES_CLOSE = 0x08, @@ -105,6 +106,16 @@ int esp_transport_ws_send_raw(esp_transport_handle_t t, ws_transport_opcodes_t o */ ws_transport_opcodes_t esp_transport_ws_get_read_opcode(esp_transport_handle_t t); +/** + * @brief Returns payload length of the last received data + * + * @param t websocket transport handle + * + * @return + * - Number of bytes in the payload + */ +int esp_transport_ws_get_read_payload_len(esp_transport_handle_t t); + #ifdef __cplusplus } diff --git a/components/tcp_transport/transport_ws.c b/components/tcp_transport/transport_ws.c index b0d0eca17..77b5d1b21 100644 --- a/components/tcp_transport/transport_ws.c +++ b/components/tcp_transport/transport_ws.c @@ -25,16 +25,24 @@ static const char *TAG = "TRANSPORT_WS"; #define WS_MASK 0x80 #define WS_SIZE16 126 #define WS_SIZE64 127 -#define MAX_WEBSOCKET_HEADER_SIZE 10 +#define MAX_WEBSOCKET_HEADER_SIZE 16 #define WS_RESPONSE_OK 101 + +typedef struct { + uint8_t opcode; + char mask_key[4]; /*!< Mask key for this payload */ + int payload_len; /*!< Total length of the payload */ + int bytes_remaining; /*!< Bytes left to read of the payload */ +} ws_transport_frame_state_t; + typedef struct { char *path; char *buffer; char *sub_protocol; char *user_agent; char *headers; - uint8_t read_opcode; + ws_transport_frame_state_t frame_state; esp_transport_handle_t parent; } transport_ws_t; @@ -46,6 +54,11 @@ static inline uint8_t ws_get_bin_opcode(ws_transport_opcodes_t opcode) static esp_transport_handle_t ws_get_payload_transport_handle(esp_transport_handle_t t) { transport_ws_t *ws = esp_transport_get_context_data(t); + + /* Reading parts of a frame directly will disrupt the WS internal frame state, + reset bytes_remaining to prepare for reading a new frame */ + ws->frame_state.bytes_remaining = 0; + return ws->parent; } @@ -89,7 +102,8 @@ static int ws_connect(esp_transport_handle_t t, const char *host, int port, int { transport_ws_t *ws = esp_transport_get_context_data(t); if (esp_transport_connect(ws->parent, host, port, timeout_ms) < 0) { - ESP_LOGE(TAG, "Error connect to ther server"); + ESP_LOGE(TAG, "Error connecting to host %s:%d", host, port); + return -1; } unsigned char random_key[16]; @@ -194,16 +208,25 @@ static int _ws_write(esp_transport_handle_t t, int opcode, int mask_flag, const ESP_LOGE(TAG, "Error transport_poll_write"); return poll_write; } - ws_header[header_len++] = opcode; - // NOTE: no support for > 16-bit sized messages - if (len > 125) { + if (len <= 125) { + ws_header[header_len++] = (uint8_t)(len | mask_flag); + } else if (len < 65536) { ws_header[header_len++] = WS_SIZE16 | mask_flag; ws_header[header_len++] = (uint8_t)(len >> 8); ws_header[header_len++] = (uint8_t)(len & 0xFF); } else { - ws_header[header_len++] = (uint8_t)(len | mask_flag); + ws_header[header_len++] = WS_SIZE64 | mask_flag; + /* Support maximum 4 bytes length */ + ws_header[header_len++] = 0; //(uint8_t)((len >> 56) & 0xFF); + ws_header[header_len++] = 0; //(uint8_t)((len >> 48) & 0xFF); + ws_header[header_len++] = 0; //(uint8_t)((len >> 40) & 0xFF); + ws_header[header_len++] = 0; //(uint8_t)((len >> 32) & 0xFF); + ws_header[header_len++] = (uint8_t)((len >> 24) & 0xFF); + ws_header[header_len++] = (uint8_t)((len >> 16) & 0xFF); + ws_header[header_len++] = (uint8_t)((len >> 8) & 0xFF); + ws_header[header_len++] = (uint8_t)((len >> 0) & 0xFF); } if (mask_flag) { @@ -215,6 +238,7 @@ static int _ws_write(esp_transport_handle_t t, int opcode, int mask_flag, const buffer[i] = (buffer[i] ^ mask[i % 4]); } } + if (esp_transport_write(ws->parent, ws_header, header_len, timeout_ms) != header_len) { ESP_LOGE(TAG, "Error write header"); return -1; @@ -252,12 +276,46 @@ static int ws_write(esp_transport_handle_t t, const char *b, int len, int timeou return _ws_write(t, WS_OPCODE_BINARY | WS_FIN, WS_MASK, b, len, timeout_ms); } -static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ms) + +static int ws_read_payload(esp_transport_handle_t t, char *buffer, int len, int timeout_ms) +{ + transport_ws_t *ws = esp_transport_get_context_data(t); + + int bytes_to_read; + int rlen = 0; + + if (ws->frame_state.bytes_remaining > len) { + ESP_LOGD(TAG, "Actual data to receive (%d) are longer than ws buffer (%d)", ws->frame_state.bytes_remaining, len); + bytes_to_read = len; + + } else { + bytes_to_read = ws->frame_state.bytes_remaining; + } + + // Receive and process payload + if (bytes_to_read != 0 && (rlen = esp_transport_read(ws->parent, buffer, bytes_to_read, timeout_ms)) <= 0) { + ESP_LOGE(TAG, "Error read data"); + return rlen; + } + ws->frame_state.bytes_remaining -= rlen; + + if (ws->frame_state.mask_key) { + for (int i = 0; i < bytes_to_read; i++) { + buffer[i] = (buffer[i] ^ ws->frame_state.mask_key[i % 4]); + } + } + return rlen; +} + + +/* Read and parse the WS header, determine length of payload */ +static int ws_read_header(esp_transport_handle_t t, char *buffer, int len, int timeout_ms) { transport_ws_t *ws = esp_transport_get_context_data(t); int payload_len; + char ws_header[MAX_WEBSOCKET_HEADER_SIZE]; - char *data_ptr = ws_header, mask, *mask_key = NULL; + char *data_ptr = ws_header, mask; int rlen; int poll_read; if ((poll_read = esp_transport_poll_read(ws->parent, timeout_ms)) <= 0) { @@ -266,16 +324,17 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ // Receive and process header first (based on header size) int header = 2; + int mask_len = 4; if ((rlen = esp_transport_read(ws->parent, data_ptr, header, timeout_ms)) <= 0) { ESP_LOGE(TAG, "Error read data"); return rlen; } - ws->read_opcode = (*data_ptr & 0x0F); + ws->frame_state.opcode = (*data_ptr & 0x0F); data_ptr ++; mask = ((*data_ptr >> 7) & 0x01); payload_len = (*data_ptr & 0x7F); data_ptr++; - ESP_LOGD(TAG, "Opcode: %d, mask: %d, len: %d\r\n", ws->read_opcode , mask, payload_len); + ESP_LOGD(TAG, "Opcode: %d, mask: %d, len: %d\r\n", ws->frame_state.opcode, mask, payload_len); if (payload_len == 126) { // headerLen += 2; if ((rlen = esp_transport_read(ws->parent, data_ptr, header, timeout_ms)) <= 0) { @@ -299,27 +358,48 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ } } - if (payload_len > len) { - ESP_LOGD(TAG, "Actual data to receive (%d) are longer than ws buffer (%d)", payload_len, len); - payload_len = len; - } - - // Then receive and process payload - if ((rlen = esp_transport_read(ws->parent, buffer, payload_len, timeout_ms)) <= 0) { - ESP_LOGE(TAG, "Error read data"); - return rlen; - } - if (mask) { - mask_key = buffer; - data_ptr = buffer + 4; - for (int i = 0; i < payload_len; i++) { - buffer[i] = (data_ptr[i] ^ mask_key[i % 4]); + // Read and store mask + if (payload_len != 0 && (rlen = esp_transport_read(ws->parent, buffer, mask_len, timeout_ms)) <= 0) { + ESP_LOGE(TAG, "Error read data"); + return rlen; } + memcpy(ws->frame_state.mask_key, buffer, mask_len); + } else { + memset(ws->frame_state.mask_key, 0, mask_len); } + + ws->frame_state.payload_len = payload_len; + ws->frame_state.bytes_remaining = payload_len; + return payload_len; } +static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_ms) +{ + int rlen = 0; + transport_ws_t *ws = esp_transport_get_context_data(t); + + // If message exceeds buffer len then subsequent reads will skip reading header and read whatever is left of the payload + if (ws->frame_state.bytes_remaining <= 0) { + if ( (rlen = ws_read_header(t, buffer, len, timeout_ms)) <= 0) { + // If something when wrong then we prepare for reading a new header + ws->frame_state.bytes_remaining = 0; + return rlen; + } + } + if (ws->frame_state.payload_len) { + if ( (rlen = ws_read_payload(t, buffer, len, timeout_ms)) <= 0) { + ESP_LOGE(TAG, "Error reading payload data"); + ws->frame_state.bytes_remaining = 0; + return rlen; + } + } + + return rlen; +} + + static int ws_poll_read(esp_transport_handle_t t, int timeout_ms) { transport_ws_t *ws = esp_transport_get_context_data(t); @@ -355,6 +435,7 @@ void esp_transport_ws_set_path(esp_transport_handle_t t, const char *path) ws->path = realloc(ws->path, strlen(path) + 1); strcpy(ws->path, path); } + esp_transport_handle_t esp_transport_ws_init(esp_transport_handle_t parent_handle) { esp_transport_handle_t t = esp_transport_init(); @@ -363,7 +444,7 @@ esp_transport_handle_t esp_transport_ws_init(esp_transport_handle_t parent_handl ws->parent = parent_handle; ws->path = strdup("/"); - ESP_TRANSPORT_MEM_CHECK(TAG, ws->path, { + ESP_TRANSPORT_MEM_CHECK(TAG, ws->path, { free(ws); return NULL; }); @@ -445,5 +526,11 @@ esp_err_t esp_transport_ws_set_headers(esp_transport_handle_t t, const char *hea ws_transport_opcodes_t esp_transport_ws_get_read_opcode(esp_transport_handle_t t) { transport_ws_t *ws = esp_transport_get_context_data(t); - return ws->read_opcode; + return ws->frame_state.opcode; +} + +int esp_transport_ws_get_read_payload_len(esp_transport_handle_t t) +{ + transport_ws_t *ws = esp_transport_get_context_data(t); + return ws->frame_state.payload_len; } \ No newline at end of file diff --git a/examples/protocols/websocket/README.md b/examples/protocols/websocket/README.md index e49bda81a..454ad376f 100644 --- a/examples/protocols/websocket/README.md +++ b/examples/protocols/websocket/README.md @@ -34,12 +34,6 @@ See the Getting Started Guide for full steps to configure and use ESP-IDF to bui ## Example Output ``` -I (482) system_api: Base MAC address is not set, read default base MAC address from BLK0 of EFUSE -I (2492) example_connect: Ethernet Link Up -I (4472) tcpip_adapter: eth ip: 192.168.2.137, mask: 255.255.255.0, gw: 192.168.2.2 -I (4472) example_connect: Connected to Ethernet -I (4472) example_connect: IPv4 address: 192.168.2.137 -I (4472) example_connect: IPv6 address: fe80:0000:0000:0000:bedd:c2ff:fed4:a92b I (4482) WEBSOCKET: Connecting to ws://echo.websocket.org... I (5012) WEBSOCKET: WEBSOCKET_EVENT_CONNECTED I (5492) WEBSOCKET: Sending hello 0000 diff --git a/examples/protocols/websocket/example_test.py b/examples/protocols/websocket/example_test.py index eeb0e6034..d861bd441 100644 --- a/examples/protocols/websocket/example_test.py +++ b/examples/protocols/websocket/example_test.py @@ -2,11 +2,15 @@ from __future__ import print_function from __future__ import unicode_literals import re import os +import sys import socket +import select import hashlib import base64 -import sys -from threading import Thread +import queue +import random +import string +from threading import Thread, Event try: import IDF @@ -20,8 +24,6 @@ except Exception: sys.path.insert(0, test_fw_path) import IDF -import DUT - def get_my_ip(): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) @@ -43,7 +45,10 @@ class Websocket: def __init__(self, port): self.port = port self.socket = socket.socket() + self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.settimeout(10.0) + self.send_q = queue.Queue() + self.shutdown = Event() def __enter__(self): try: @@ -56,23 +61,27 @@ class Websocket: self.server_thread = Thread(target=self.run_server) self.server_thread.start() + return self + def __exit__(self, exc_type, exc_value, traceback): + self.shutdown.set() self.server_thread.join() self.socket.close() self.conn.close() def run_server(self): self.conn, address = self.socket.accept() # accept new connection - self.conn.settimeout(10.0) + self.socket.settimeout(10.0) + print("Connection from: {}".format(address)) self.establish_connection() - - # Echo data until client closes connection - self.echo_data() + print("WS established") + # Handle connection until client closes it, will echo any data received and send data from send_q queue + self.handle_conn() def establish_connection(self): - while True: + while not self.shutdown.is_set(): try: # receive data stream. it won't accept data packet greater than 1024 bytes data = self.conn.recv(1024).decode() @@ -83,6 +92,7 @@ class Websocket: if "Upgrade: websocket" in data and "Connection: Upgrade" in data: self.handshake(data) return + except socket.error as err: print("Unable to establish a websocket connection: {}, {}".format(err)) raise @@ -107,26 +117,46 @@ class Websocket: self.conn.send(resp.encode()) - def echo_data(self): - while(True): + def handle_conn(self): + while not self.shutdown.is_set(): + r,w,e = select.select([self.conn], [], [], 1) try: - header = bytearray(self.conn.recv(self.HEADER_LEN, socket.MSG_WAITALL)) - if not header: - # exit if data is not received - return + if self.conn in r: + self.echo_data() - # Remove mask bit - payload_len = ~(1 << 7) & header[1] + if not self.send_q.empty(): + self._send_data_(self.send_q.get()) - payload = bytearray(self.conn.recv(payload_len, socket.MSG_WAITALL)) - frame = header + payload - - decoded_payload = self.decode_frame(frame) - - echo_frame = self.encode_frame(decoded_payload) - self.conn.send(echo_frame) except socket.error as err: print("Stopped echoing data: {}".format(err)) + raise + + def echo_data(self): + header = bytearray(self.conn.recv(self.HEADER_LEN, socket.MSG_WAITALL)) + if not header: + # exit if socket closed by peer + return + + # Remove mask bit + payload_len = ~(1 << 7) & header[1] + + payload = bytearray(self.conn.recv(payload_len, socket.MSG_WAITALL)) + + if not payload: + # exit if socket closed by peer + return + frame = header + payload + + decoded_payload = self.decode_frame(frame) + print("Sending echo...") + self._send_data_(decoded_payload) + + def _send_data_(self, data): + frame = self.encode_frame(data) + self.conn.send(frame) + + def send_data(self, data): + self.send_q.put(data.encode()) def decode_frame(self, frame): # Mask out MASK bit from payload length, this len is only valid for short messages (<126) @@ -147,7 +177,17 @@ class Websocket: header = (1 << 7) | (1 << 0) frame = bytearray([header]) - frame.append(len(payload)) + payload_len = len(payload) + + # If payload len is longer than 125 then the next 16 bits are used to encode length + if payload_len > 125: + frame.append(126) + frame.append(payload_len >> 8) + frame.append(0xFF & payload_len) + + else: + frame.append(payload_len) + frame += payload return frame @@ -156,8 +196,27 @@ class Websocket: def test_echo(dut): dut.expect("WEBSOCKET_EVENT_CONNECTED") for i in range(0, 10): - dut.expect(re.compile(r"Received=hello (\d)")) - dut.expect("Websocket Stopped") + dut.expect(re.compile(r"Received=hello (\d)"), timeout=30) + print("All echos received") + + +def test_recv_long_msg(dut, websocket, msg_len, repeats): + send_msg = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(msg_len)) + + for _ in range(repeats): + websocket.send_data(send_msg) + + recv_msg = '' + while len(recv_msg) < msg_len: + # Filter out color encoding + match = dut.expect(re.compile(r"Received=([a-zA-Z0-9]*).*\n"), timeout=30)[0] + recv_msg += match + + if recv_msg == send_msg: + print("Sent message and received message are equal") + else: + raise ValueError("DUT received string do not match sent string, \nexpected: {}\nwith length {}\ + \nreceived: {}\nwith length {}".format(send_msg, len(send_msg), recv_msg, len(recv_msg))) @IDF.idf_example_test(env_tag="Example_WIFI") @@ -191,12 +250,14 @@ def test_examples_protocol_websocket(env, extra_data): if uri_from_stdin: server_port = 4455 - with Websocket(server_port): + with Websocket(server_port) as ws: uri = "ws://{}:{}".format(get_my_ip(), server_port) print("DUT connecting to {}".format(uri)) dut1.expect("Please enter uri of websocket endpoint", timeout=30) dut1.write(uri) test_echo(dut1) + # Message length should exceed DUT's buffer size to test fragmentation, default is 1024 byte + test_recv_long_msg(dut1, ws, 2000, 3) else: print("DUT connecting to {}".format(uri)) diff --git a/examples/protocols/websocket/main/websocket_example.c b/examples/protocols/websocket/main/websocket_example.c index d74474914..8b4bf8600 100644 --- a/examples/protocols/websocket/main/websocket_example.c +++ b/examples/protocols/websocket/main/websocket_example.c @@ -1,4 +1,4 @@ -/* ESP Websocket Client Example +/* ESP HTTP Client Example This example code is in the Public Domain (or CC0 licensed, at your option.) @@ -16,6 +16,7 @@ #include "freertos/FreeRTOS.h" #include "freertos/task.h" +#include "freertos/semphr.h" #include "freertos/event_groups.h" @@ -23,11 +24,22 @@ #include "esp_websocket_client.h" #include "esp_event.h" +#define NO_DATA_TIMEOUT_SEC 10 + static const char *TAG = "WEBSOCKET"; static EventGroupHandle_t wifi_event_group; const static int CONNECTED_BIT = BIT0; +static TimerHandle_t shutdown_signal_timer; +static SemaphoreHandle_t shutdown_sema; + +static void shutdown_signaler(TimerHandle_t xTimer) +{ + ESP_LOGI(TAG, "No data received for %d seconds, signaling shutdown", NO_DATA_TIMEOUT_SEC); + xSemaphoreGive(shutdown_sema); +} + #if CONFIG_WEBSOCKET_URI_FROM_STDIN static void get_string(char *line, size_t size) { @@ -51,20 +63,23 @@ static void websocket_event_handler(void *handler_args, esp_event_base_t base, i { esp_websocket_event_data_t *data = (esp_websocket_event_data_t *)event_data; switch (event_id) { - case WEBSOCKET_EVENT_CONNECTED: - ESP_LOGI(TAG, "WEBSOCKET_EVENT_CONNECTED"); - break; - case WEBSOCKET_EVENT_DISCONNECTED: - ESP_LOGI(TAG, "WEBSOCKET_EVENT_DISCONNECTED"); - break; - case WEBSOCKET_EVENT_DATA: - ESP_LOGI(TAG, "WEBSOCKET_EVENT_DATA"); - ESP_LOGI(TAG, "Received opcode=%d", data->op_code); - ESP_LOGW(TAG, "Received=%.*s\r\n", data->data_len, (char*)data->data_ptr); - break; - case WEBSOCKET_EVENT_ERROR: - ESP_LOGI(TAG, "WEBSOCKET_EVENT_ERROR"); - break; + case WEBSOCKET_EVENT_CONNECTED: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_CONNECTED"); + break; + case WEBSOCKET_EVENT_DISCONNECTED: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_DISCONNECTED"); + break; + case WEBSOCKET_EVENT_DATA: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_DATA"); + ESP_LOGI(TAG, "Received opcode=%d", data->op_code); + ESP_LOGW(TAG, "Received=%.*s", data->data_len, (char *)data->data_ptr); + ESP_LOGW(TAG, "Total payload length=%d, data_len=%d, current payload offset=%d\r\n", data->payload_len, data->data_len, data->payload_offset); + + xTimerReset(shutdown_signal_timer, portMAX_DELAY); + break; + case WEBSOCKET_EVENT_ERROR: + ESP_LOGI(TAG, "WEBSOCKET_EVENT_ERROR"); + break; } } @@ -114,7 +129,11 @@ static void websocket_app_start(void) { esp_websocket_client_config_t websocket_cfg = {}; - #if CONFIG_WEBSOCKET_URI_FROM_STDIN + shutdown_signal_timer = xTimerCreate("Websocket shutdown timer", NO_DATA_TIMEOUT_SEC * 1000 / portTICK_PERIOD_MS, + pdFALSE, NULL, shutdown_signaler); + shutdown_sema = xSemaphoreCreateBinary(); + +#if CONFIG_WEBSOCKET_URI_FROM_STDIN char line[128]; ESP_LOGI(TAG, "Please enter uri of websocket endpoint"); @@ -123,10 +142,10 @@ static void websocket_app_start(void) websocket_cfg.uri = line; ESP_LOGI(TAG, "Endpoint uri: %s\n", line); - #else +#else websocket_cfg.uri = CONFIG_WEBSOCKET_URI; - #endif /* CONFIG_WEBSOCKET_URI_FROM_STDIN */ +#endif /* CONFIG_WEBSOCKET_URI_FROM_STDIN */ ESP_LOGI(TAG, "Connecting to %s...", websocket_cfg.uri); @@ -134,6 +153,7 @@ static void websocket_app_start(void) esp_websocket_register_events(client, WEBSOCKET_EVENT_ANY, websocket_event_handler, (void *)client); esp_websocket_client_start(client); + xTimerStart(shutdown_signal_timer, portMAX_DELAY); char data[32]; int i = 0; while (i < 10) { @@ -144,8 +164,8 @@ static void websocket_app_start(void) } vTaskDelay(1000 / portTICK_RATE_MS); } - // Give server some time to respond before closing - vTaskDelay(3000 / portTICK_RATE_MS); + + xSemaphoreTake(shutdown_sema, portMAX_DELAY); esp_websocket_client_stop(client); ESP_LOGI(TAG, "Websocket Stopped"); esp_websocket_client_destroy(client); @@ -164,4 +184,4 @@ void app_main(void) nvs_flash_init(); wifi_init(); websocket_app_start(); -} \ No newline at end of file +}