Skip to content

Commit

Permalink
feat(websocket): Added new APIs to support fragmented messages transm…
Browse files Browse the repository at this point in the history 
…ission

Intoduced new APIs`esp_websocket_client_send_text_partial`,
            `esp_websocket_client_send_bin_partial`
            `esp_websocket_client_send_cont_mgs`
            `esp_websocket_client_send_fin`
            `esp_websocket_client_send_with_exact_opcode`
  • Loading branch information
@suren-gabrielyan-espressif
suren-gabrielyan-espressif committed Oct 3, 2023
1 parent 0db29f2 commit f97eed0
Show file tree
Hide file tree
Showing 4 changed files with 144 additions and 38 deletions.
98 changes: 61 additions & 37 deletions components/esp_websocket_client/esp_websocket_client.c
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* SPDX-FileCopyrightText: 2015-2022 Espressif Systems (Shanghai) CO LTD
* SPDX-FileCopyrightText: 2015-2023 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
Expand Down Expand Up @@ -545,6 +545,42 @@ static esp_err_t esp_websocket_client_create_transport(esp_websocket_client_hand
return ESP_OK;
}

static bool esp_websocket_client_send_with_exact_opcode(esp_websocket_client_handle_t client, ws_transport_opcodes_t opcode, const uint8_t *data, int len, TickType_t timeout)
{
int ret = -1;
int need_write = len;
int wlen = 0, widx = 0;

while (widx < len || opcode) { // allow for sending "current_opcode" only message with len==0
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 || (wlen == 0 && need_write != 0)) {
ret = wlen;
esp_websocket_free_buf(client, true);
esp_tls_error_handle_t error_handle = esp_transport_get_error_handle(client->transport);
if (error_handle) {
esp_websocket_client_error(client, "esp_transport_write() returned %d, transport_error=%s, tls_error_code=%i, tls_flags=%i, errno=%d",
ret, esp_err_to_name(error_handle->last_error), error_handle->esp_tls_error_code,
error_handle->esp_tls_flags, errno);
} else {
esp_websocket_client_error(client, "esp_transport_write() returned %d, errno=%d", ret, errno);
}
esp_websocket_client_abort_connection(client, WEBSOCKET_ERROR_TYPE_TCP_TRANSPORT);
return false;
}
opcode = 0;
widx += wlen;
need_write = len - widx;
}
esp_websocket_free_buf(client, true);
return true;
}

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));
Expand Down Expand Up @@ -1092,17 +1128,33 @@ int esp_websocket_client_send_text(esp_websocket_client_handle_t client, const c
return esp_websocket_client_send_with_opcode(client, WS_TRANSPORT_OPCODES_TEXT, (const uint8_t *)data, len, timeout);
}

int esp_websocket_client_send_text_partial(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout)
{
return esp_websocket_client_send_with_exact_opcode(client, WS_TRANSPORT_OPCODES_TEXT, (const uint8_t *)data, len, timeout);
}

int esp_websocket_client_send_cont_msg(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout)
{
return esp_websocket_client_send_with_exact_opcode(client, WS_TRANSPORT_OPCODES_CONT, (const uint8_t *)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, (const uint8_t *)data, len, timeout);
}

int esp_websocket_client_send_with_opcode(esp_websocket_client_handle_t client, ws_transport_opcodes_t opcode, const uint8_t *data, int len, TickType_t timeout)
int esp_websocket_client_send_bin_partial(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout)
{
int need_write = len;
int wlen = 0, widx = 0;
int ret = ESP_FAIL;
return esp_websocket_client_send_with_exact_opcode(client, WS_TRANSPORT_OPCODES_BINARY, (const uint8_t *)data, len, timeout);
}

int esp_websocket_client_send_fin(esp_websocket_client_handle_t client, TickType_t timeout)
{
return esp_websocket_client_send_with_exact_opcode(client, WS_TRANSPORT_OPCODES_FIN, NULL, 0, timeout);
}

int esp_websocket_client_send_with_opcode(esp_websocket_client_handle_t client, ws_transport_opcodes_t opcode, const uint8_t *data, int len, TickType_t timeout)
{
if (client == NULL || len < 0 || (data == NULL && len > 0)) {
ESP_LOGE(TAG, "Invalid arguments");
return ESP_FAIL;
Expand All @@ -1126,41 +1178,13 @@ int esp_websocket_client_send_with_opcode(esp_websocket_client_handle_t client,
ESP_LOGE(TAG, "Failed to setup tx buffer");
goto unlock_and_return;
}
uint32_t current_opcode = opcode;
while (widx < len || current_opcode) { // allow for sending "current_opcode" only message with len==0
if (need_write > client->buffer_size) {
need_write = client->buffer_size;
} else {
current_opcode |= WS_TRANSPORT_OPCODES_FIN;
}
memcpy(client->tx_buffer, data + widx, need_write);
// send with ws specific way and specific opcode
wlen = esp_transport_ws_send_raw(client->transport, current_opcode, (char *)client->tx_buffer, need_write,
(timeout == portMAX_DELAY) ? -1 : timeout * portTICK_PERIOD_MS);
if (wlen < 0 || (wlen == 0 && need_write != 0)) {
ret = wlen;
esp_websocket_free_buf(client, true);
esp_tls_error_handle_t error_handle = esp_transport_get_error_handle(client->transport);
if (error_handle) {
esp_websocket_client_error(client, "esp_transport_write() returned %d, transport_error=%s, tls_error_code=%i, tls_flags=%i, errno=%d",
ret, esp_err_to_name(error_handle->last_error), error_handle->esp_tls_error_code,
error_handle->esp_tls_flags, errno);
} else {
esp_websocket_client_error(client, "esp_transport_write() returned %d, errno=%d", ret, errno);
}
esp_websocket_client_abort_connection(client, WEBSOCKET_ERROR_TYPE_TCP_TRANSPORT);
goto unlock_and_return;
}
current_opcode = 0;
widx += wlen;
need_write = len - widx;

if (esp_websocket_client_send_with_exact_opcode(client, opcode | WS_TRANSPORT_OPCODES_FIN, data, len, timeout) != true) {
ESP_LOGE(TAG, "Failed to send the buffer");
goto unlock_and_return;
}
ret = widx;
esp_websocket_free_buf(client, true);
unlock_and_return:
xSemaphoreGiveRecursive(client->lock);
return ret;
return ESP_FAIL;
}

bool esp_websocket_client_is_connected(esp_websocket_client_handle_t client)
Expand Down
8 changes: 8 additions & 0 deletions components/esp_websocket_client/examples/target/main/websocket_example.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,14 @@ static void websocket_app_start(void)
vTaskDelay(1000 / portTICK_PERIOD_MS);
}

ESP_LOGI(TAG, "Sending fragmented message");
memset(data, 'a', sizeof(data));
esp_websocket_client_send_text_partial(client, data, sizeof(data), portMAX_DELAY);
memset(data, 'b', sizeof(data));
esp_websocket_client_send_cont_msg(client, data, sizeof(data), portMAX_DELAY);
esp_websocket_client_send_fin(client, portMAX_DELAY);
vTaskDelay(1000 / portTICK_PERIOD_MS);

xSemaphoreTake(shutdown_sema, portMAX_DELAY);
esp_websocket_client_close(client, portMAX_DELAY);
ESP_LOGI(TAG, "Websocket Stopped");
Expand Down
5 changes: 5 additions & 0 deletions components/esp_websocket_client/examples/target/pytest_websocket.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,10 @@ def test_recv_long_msg(dut, websocket, msg_len, repeats):
\nreceived: {}\nwith length {}'.format(
send_msg, len(send_msg), recv_msg, len(recv_msg)))

def test_fragmented_msg(dut):
dut.expect('Received=' + 32 * 'a' + 32 * 'b')
print('Fragmented data received')

# Starting of the test
try:
if dut.app.sdkconfig.get('WEBSOCKET_URI_FROM_STDIN') is True:
Expand All @@ -156,6 +160,7 @@ def test_recv_long_msg(dut, websocket, msg_len, repeats):
# Message length should exceed DUT's buffer size to test fragmentation, default is 1024 byte
test_recv_long_msg(dut, ws, 2000, 3)
test_json(dut, ws)
test_fragmented_msg(dut)
test_close(dut)
else:
print('DUT connecting to {}'.format(uri))
Expand Down
71 changes: 70 additions & 1 deletion components/esp_websocket_client/include/esp_websocket_client.h
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* SPDX-FileCopyrightText: 2015-2021 Espressif Systems (Shanghai) CO LTD
* SPDX-FileCopyrightText: 2015-2023 Espressif Systems (Shanghai) CO LTD
*
* SPDX-License-Identifier: Apache-2.0
*/
Expand Down Expand Up @@ -231,6 +231,24 @@ esp_err_t esp_websocket_client_destroy_on_exit(esp_websocket_client_handle_t cli
*/
int esp_websocket_client_send_bin(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout);

/**
* @brief Write binary data to the WebSocket connection and sends it without setting the FIN flag(data send with WS OPCODE=02, i.e. binary)
*
* Notes:
* - To send continuation frame, you should use 'esp_websocket_client_send_cont_msg(...)' API.
* - To mark the end of fragmented data, you should use the 'esp_websocket_client_send_fin(...)' API. This sends a FIN frame.
*
* @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_partial(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)
*
Expand All @@ -245,6 +263,55 @@ int esp_websocket_client_send_bin(esp_websocket_client_handle_t client, const ch
*/
int esp_websocket_client_send_text(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout);

/**
* @brief Write textual data to the WebSocket connection and sends it without setting the FIN flag(data send with WS OPCODE=01, i.e. text)
*
* Notes:
* - To send continuation frame, you should use 'esp_websocket_client_send_cont_mgs(...)' API.
* - To mark the end of fragmented data, you should use the 'esp_websocket_client_send_fin(...)' API. This sends a FIN frame.
*
* @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_partial(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout);

/**
* @brief Write textual data to the WebSocket connection and sends it as continuation frame (OPCODE=0x0)
*
* Notes:
* - Continuation frames have an opcode of 0x0 and do not explicitly signify whether they are continuing a text or a binary message.
* - You determine the type of message (text or binary) being continued by looking at the opcode of the initial frame in the sequence of fragmented frames.
* - To mark the end of fragmented data, you should use the 'esp_websocket_client_send_fin(...)' API. This sends a FIN frame.
*
* @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_cont_msg(esp_websocket_client_handle_t client, const char *data, int len, TickType_t timeout);

/**
* @brief Sends FIN frame
*
* @param[in] client The client
* @param[in] timeout Write data timeout in RTOS ticks
*
* @return
* - Number of data was sent
* - (-1) if any errors
*/
int esp_websocket_client_send_fin(esp_websocket_client_handle_t client, TickType_t timeout);

/**
* @brief Write opcode data to the WebSocket connection
*
Expand All @@ -256,6 +323,8 @@ int esp_websocket_client_send_text(esp_websocket_client_handle_t client, const c
*
* Notes:
* - In order to send a zero payload, data and len should be set to NULL/0
* - This API sets the FIN bit on the last fragment of message
*
*
* @return
* - Number of data was sent
Expand Down

0 comments on commit f97eed0

Please sign in to comment.