espressif/esp-mqtt
1
0
Fork 0
mirror of https://github.com/espressif/esp-mqtt.git synced 2026-10-21 23:18:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Move inbound message data to be managed by message type

Browse Source 
- Incoming path was using just buffers this moves the buffers to be
managed as message.
This commit is contained in:
Euripedes Rocha
2026-10-01 06:53:56 +02:00
committed by Euripedes Rocha Filho
parent 313168f39c
commit b39bed5fb8
5 changed files with 98 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/include/mqtt_client_priv.h
+1 -4
View File
@@ -62,11 +62,8 @@ typedef struct mqtt_connection {
} mqtt_connection_t;
typedef struct mqtt_state {
uint8_t *in_buffer;
int in_buffer_length;
size_t message_length;
size_t in_buffer_read_len;
mqtt_connection_t connection;
mqtt_message_t inbound_message;
} mqtt_state_t;
typedef struct {
lib/include/mqtt_msg.h
+1
View File
@@ -44,6 +44,7 @@ typedef struct mqtt_message {
uint8_t *data;
uint8_t *buffer;
size_t buffer_length;
size_t read_len;
uint16_t id;
uint16_t last_message_id; /*!< last generated id when incremental message ids are configured */
int type;
lib/mqtt_msg.c
+1
View File
@@ -639,6 +639,7 @@ esp_err_t mqtt_msg_buffer_init(mqtt_message_t *message, int buffer_size)
return ESP_ERR_NO_MEM;
}
message->data = message->buffer;
message->buffer_length = buffer_size;
return ESP_OK;
}
mqtt5_client.c
+23 -23
View File
@@ -79,11 +79,11 @@ void esp_mqtt5_decrement_packet_counter(esp_mqtt5_client_handle_t client)
void esp_mqtt5_parse_pubcomp(esp_mqtt5_client_handle_t client)
{
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PUBCOMP return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len);
client->event.data = mqtt5_get_pubcomp_data(client->mqtt_state.in_buffer, &msg_data_len,
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PUBCOMP return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data,
client->mqtt_state.inbound_message.read_len));
size_t msg_data_len = client->mqtt_state.inbound_message.read_len;
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.read_len);
client->event.data = mqtt5_get_pubcomp_data(client->mqtt_state.inbound_message.data, &msg_data_len,
&client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
@@ -94,11 +94,11 @@ void esp_mqtt5_parse_pubcomp(esp_mqtt5_client_handle_t client)
void esp_mqtt5_parse_puback(esp_mqtt5_client_handle_t client)
{
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len);
client->event.data = mqtt5_get_puback_data(client->mqtt_state.in_buffer, &msg_data_len,
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data,
client->mqtt_state.inbound_message.read_len));
size_t msg_data_len = client->mqtt_state.inbound_message.read_len;
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.read_len);
client->event.data = mqtt5_get_puback_data(client->mqtt_state.inbound_message.data, &msg_data_len,
&client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
@@ -109,11 +109,11 @@ void esp_mqtt5_parse_puback(esp_mqtt5_client_handle_t client)
void esp_mqtt5_parse_unsuback(esp_mqtt5_client_handle_t client)
{
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGD(TAG, "MQTT_MSG_TYPE_UNSUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_read_len));
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len);
client->event.data = mqtt5_get_unsuback_data(client->mqtt_state.in_buffer, &msg_data_len,
ESP_LOGD(TAG, "MQTT_MSG_TYPE_UNSUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data,
client->mqtt_state.inbound_message.read_len));
size_t msg_data_len = client->mqtt_state.inbound_message.read_len;
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.read_len);
client->event.data = mqtt5_get_unsuback_data(client->mqtt_state.inbound_message.data, &msg_data_len,
&client->event.property->user_property);
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
@@ -124,28 +124,28 @@ void esp_mqtt5_parse_unsuback(esp_mqtt5_client_handle_t client)
void esp_mqtt5_parse_suback(esp_mqtt5_client_handle_t client)
{
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
ESP_LOGD(TAG, "MQTT_MSG_TYPE_SUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_read_len));
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len);
ESP_LOGD(TAG, "MQTT_MSG_TYPE_SUBACK return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data,
client->mqtt_state.inbound_message.read_len));
client->event.reason_code = mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.read_len);
}
}
void esp_mqtt5_parse_disconnect(esp_mqtt5_client_handle_t client, int *disconnect_rsp_code)
{
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
*disconnect_rsp_code = mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len);
*disconnect_rsp_code = mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.read_len);
ESP_LOGD(TAG, "MQTT_MSG_TYPE_DISCONNECT return code is %d", *disconnect_rsp_code);
}
}
esp_err_t esp_mqtt5_parse_connack(esp_mqtt5_client_handle_t client, int *connect_rsp_code)
{
size_t len = client->mqtt_state.in_buffer_read_len;
client->mqtt_state.in_buffer_read_len = 0;
size_t len = client->mqtt_state.inbound_message.read_len;
client->mqtt_state.inbound_message.read_len = 0;
uint8_t ack_flag = 0;
client->mqtt5_config->server_resp_property_info.receive_maximum = MQTT5_DEFAULT_RECEIVE_MAXIMUM;
if (mqtt5_msg_parse_connack_property(client->mqtt_state.in_buffer, len, &client->mqtt_state.
if (mqtt5_msg_parse_connack_property(client->mqtt_state.inbound_message.data, len, &client->mqtt_state.
connection.information, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->server_resp_property_info,
connect_rsp_code, &ack_flag, &client->event.property->user_property) != ESP_OK) {
ESP_LOGE(TAG, "Failed to parse CONNACK packet");
@@ -735,7 +735,7 @@ esp_err_t esp_mqtt5_client_set_connect_property(esp_mqtt5_client_handle_t client
if (connect_property->maximum_packet_size) {
client->mqtt5_config->connect_property_info.maximum_packet_size = connect_property->maximum_packet_size;
} else {
client->mqtt5_config->connect_property_info.maximum_packet_size = client->mqtt_state.in_buffer_length;
client->mqtt5_config->connect_property_info.maximum_packet_size = client->mqtt_state.inbound_message.buffer_length;
}
if (connect_property->receive_maximum) {
mqtt_client.c
+72 -70
View File
@@ -438,10 +438,12 @@ esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_cl
}
client->mqtt_state.connection.outbound_message.last_message_id = last_message_id;
free(client->mqtt_state.in_buffer);
client->mqtt_state.in_buffer = (uint8_t *)heap_caps_malloc(buffer_size, MQTT_BUFFER_MEMORY);
ESP_MEM_CHECK(TAG, client->mqtt_state.in_buffer, goto _mqtt_set_config_failed);
client->mqtt_state.in_buffer_length = buffer_size;
mqtt_msg_buffer_destroy(&client->mqtt_state.inbound_message);
if (mqtt_msg_buffer_init(&client->mqtt_state.inbound_message, buffer_size) != ESP_OK) {
goto _mqtt_set_config_failed;
}
client->config->message_retransmit_timeout = config->session.message_retransmit_timeout;
if (config->session.message_retransmit_timeout <= 0) {
@@ -693,7 +695,7 @@ void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
return;
}
free(client->mqtt_state.in_buffer);
mqtt_msg_buffer_destroy(&client->mqtt_state.inbound_message);
mqtt_msg_buffer_destroy(&client->mqtt_state.connection.outbound_message);
free(client->config->host);
free(client->config->uri);
@@ -872,8 +874,8 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
return ESP_FAIL;
}
client->mqtt_state.in_buffer_read_len = 0;
client->mqtt_state.message_length = 0;
client->mqtt_state.inbound_message.read_len = 0;
client->mqtt_state.inbound_message.length = 0;
/* wait configured network timeout for broker connection response */
uint64_t connack_recv_started = platform_tick_get_ms();
@@ -886,8 +888,8 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
return ESP_FAIL;
}
if (mqtt_get_type(client->mqtt_state.in_buffer) != MQTT_MSG_TYPE_CONNACK) {
ESP_LOGE(TAG, "Invalid MSG_TYPE response: %d, read_len: %d", mqtt_get_type(client->mqtt_state.in_buffer), read_len);
if (mqtt_get_type(client->mqtt_state.inbound_message.data) != MQTT_MSG_TYPE_CONNACK) {
ESP_LOGE(TAG, "Invalid MSG_TYPE response: %d, read_len: %d", mqtt_get_type(client->mqtt_state.inbound_message.data), read_len);
return ESP_FAIL;
}
@@ -902,8 +904,8 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
#endif
} else {
client->mqtt_state.in_buffer_read_len = 0;
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
client->mqtt_state.inbound_message.read_len = 0;
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.inbound_message.data);
if (connect_rsp_code == MQTT_CONNECTION_ACCEPTED) {
ESP_LOGD(TAG, "Connected");
@@ -1190,10 +1192,10 @@ static esp_err_t esp_mqtt_dispatch_event_with_msgid(esp_mqtt_client_handle_t cli
{
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->event.msg_id = mqtt5_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
client->event.msg_id = mqtt5_get_id(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.buffer_length);
#endif
} else {
client->event.msg_id = mqtt_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
client->event.msg_id = mqtt_get_id(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.buffer_length);
}
return esp_mqtt_dispatch_event(client);
@@ -1242,9 +1244,9 @@ static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client)
static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
{
uint8_t *msg_buf = client->mqtt_state.in_buffer;
size_t msg_read_len = client->mqtt_state.in_buffer_read_len;
size_t msg_total_len = client->mqtt_state.message_length;
uint8_t *msg_buf = client->mqtt_state.inbound_message.data;
size_t msg_read_len = client->mqtt_state.inbound_message.read_len;
size_t msg_total_len = client->mqtt_state.inbound_message.length;
size_t msg_topic_len = msg_read_len;
size_t msg_data_len = msg_read_len;
size_t msg_data_offset = 0;
@@ -1321,12 +1323,12 @@ static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
}
#endif
size_t buf_len = client->mqtt_state.in_buffer_length;
msg_data = (char *)client->mqtt_state.in_buffer;
size_t buf_len = client->mqtt_state.inbound_message.buffer_length;
msg_data = (char *)client->mqtt_state.inbound_message.data;
msg_topic = saved_msg_topic;
msg_topic_len = saved_msg_topic_len;
msg_data_offset += msg_data_len;
int ret = esp_transport_read(client->transport.handle, (char *)client->mqtt_state.in_buffer,
int ret = esp_transport_read(client->transport.handle, (char *)client->mqtt_state.inbound_message.data,
msg_total_len - msg_read_len > buf_len ? buf_len : msg_total_len - msg_read_len,
client->config->network_timeout_ms);
@@ -1345,8 +1347,8 @@ static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
static esp_err_t deliver_suback(esp_mqtt_client_handle_t client)
{
uint8_t *msg_buf = client->mqtt_state.in_buffer;
size_t msg_data_len = client->mqtt_state.in_buffer_read_len;
uint8_t *msg_buf = client->mqtt_state.inbound_message.data;
size_t msg_data_len = client->mqtt_state.inbound_message.read_len;
char *msg_data = NULL;
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
@@ -1423,18 +1425,18 @@ static outbox_item_handle_t mqtt_enqueue(esp_mqtt_client_handle_t client, const
* -1 timeout while in-the-middle of the message
* 0 if no message has been received
* 1 if a message has been received and placed to client->mqtt_state:
* message length: client->mqtt_state.message_length
* message content: client->mqtt_state.in_buffer
* message length: client->mqtt_state.inbound_message.length
* message content: client->mqtt_state.inbound_message.data
*
*/
static int mqtt_message_receive(esp_mqtt_client_handle_t client, int read_poll_timeout_ms)
{
int read_len, total_len, fixed_header_len;
uint8_t *buf = client->mqtt_state.in_buffer + client->mqtt_state.in_buffer_read_len;
uint8_t *buf = client->mqtt_state.inbound_message.data + client->mqtt_state.inbound_message.read_len;
esp_transport_handle_t t = client->transport.handle;
client->mqtt_state.message_length = 0;
client->mqtt_state.inbound_message.length = 0;
if (client->mqtt_state.in_buffer_read_len == 0) {
if (client->mqtt_state.inbound_message.read_len == 0) {
/*
* Read first byte of the mqtt packet fixed header, it contains packet
* type and flags.
@@ -1457,11 +1459,11 @@ static int mqtt_message_receive(esp_mqtt_client_handle_t client, int read_poll_t
}
buf++;
client->mqtt_state.in_buffer_read_len++;
client->mqtt_state.inbound_message.read_len++;
}
if ((client->mqtt_state.in_buffer_read_len == 1) ||
((client->mqtt_state.in_buffer_read_len < 6) && (*(buf - 1) & 0x80))) {
if ((client->mqtt_state.inbound_message.read_len == 1) ||
((client->mqtt_state.inbound_message.read_len < 6) && (*(buf - 1) & 0x80))) {
do {
/*
* Read the "remaining length" part of mqtt packet fixed header. It
@@ -1478,25 +1480,25 @@ static int mqtt_message_receive(esp_mqtt_client_handle_t client, int read_poll_t
ESP_LOGD(TAG, "%s: read \"remaining length\" byte: 0x%x", __func__, *buf);
buf++;
client->mqtt_state.in_buffer_read_len++;
} while ((client->mqtt_state.in_buffer_read_len < 6) && (*(buf - 1) & 0x80));
client->mqtt_state.inbound_message.read_len++;
} while ((client->mqtt_state.inbound_message.read_len < 6) && (*(buf - 1) & 0x80));
}
total_len = mqtt_get_total_length(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len,
total_len = mqtt_get_total_length(client->mqtt_state.inbound_message.data, client->mqtt_state.inbound_message.read_len,
&fixed_header_len);
ESP_LOGD(TAG, "%s: total message length: %d (already read: %"NEWLIB_NANO_COMPAT_FORMAT")", __func__, total_len,
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.in_buffer_read_len));
client->mqtt_state.message_length = total_len;
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.read_len));
client->mqtt_state.inbound_message.length = total_len;
if (client->mqtt_state.in_buffer_length < total_len) {
if (mqtt_get_type(client->mqtt_state.in_buffer) == MQTT_MSG_TYPE_PUBLISH) {
if (client->mqtt_state.inbound_message.buffer_length < total_len) {
if (mqtt_get_type(client->mqtt_state.inbound_message.data) == MQTT_MSG_TYPE_PUBLISH) {
/*
* In case larger publish messages, we only need to read full topic, data can be split to multiple data event.
* Evaluate and correct total_len to read only publish message header, so data can be read separately
*/
if (client->mqtt_state.in_buffer_read_len < fixed_header_len + 2) {
if (client->mqtt_state.inbound_message.read_len < fixed_header_len + 2) {
/* read next 2 bytes - topic length to get minimum portion of publish packet */
read_len = esp_transport_read(t, (char *)buf, client->mqtt_state.in_buffer_read_len - fixed_header_len + 2,
read_len = esp_transport_read(t, (char *)buf, client->mqtt_state.inbound_message.read_len - fixed_header_len + 2,
read_poll_timeout_ms);
ESP_LOGD(TAG, "%s: read_len=%d", __func__, read_len);
@@ -1504,27 +1506,27 @@ static int mqtt_message_receive(esp_mqtt_client_handle_t client, int read_poll_t
return esp_mqtt_handle_transport_read_error(read_len, client, true);
}
client->mqtt_state.in_buffer_read_len += read_len;
client->mqtt_state.inbound_message.read_len += read_len;
buf += read_len;
if (client->mqtt_state.in_buffer_read_len < fixed_header_len + 2) {
if (client->mqtt_state.inbound_message.read_len < fixed_header_len + 2) {
ESP_LOGD(TAG,
"%s: transport_read(): message reading left in progress :: total message length: %d (already read: %"NEWLIB_NANO_COMPAT_FORMAT")",
__func__, total_len, NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.in_buffer_read_len));
__func__, total_len, NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.read_len));
return 0;
}
}
int topic_len = client->mqtt_state.in_buffer[fixed_header_len] << 8;
topic_len |= client->mqtt_state.in_buffer[fixed_header_len + 1];
total_len = fixed_header_len + topic_len + (mqtt_get_qos(client->mqtt_state.in_buffer) > 0 ? 2 : 0);
int topic_len = client->mqtt_state.inbound_message.data[fixed_header_len] << 8;
topic_len |= client->mqtt_state.inbound_message.data[fixed_header_len + 1];
total_len = fixed_header_len + topic_len + (mqtt_get_qos(client->mqtt_state.inbound_message.data) > 0 ? 2 : 0);
ESP_LOGD(TAG, "%s: total len modified to %d as message longer than input buffer", __func__, total_len);
if (client->mqtt_state.in_buffer_length < total_len) {
if (client->mqtt_state.inbound_message.buffer_length < total_len) {
ESP_LOGE(TAG, "%s: message is too big, insufficient buffer size", __func__);
goto err;
} else {
total_len = client->mqtt_state.in_buffer_length;
total_len = client->mqtt_state.inbound_message.buffer_length;
}
/* free to continue with reading */
@@ -1534,28 +1536,28 @@ static int mqtt_message_receive(esp_mqtt_client_handle_t client, int read_poll_t
}
}
if (client->mqtt_state.in_buffer_read_len < total_len) {
if (client->mqtt_state.inbound_message.read_len < total_len) {
/* read the rest of the mqtt message */
read_len = esp_transport_read(t, (char *)buf, total_len - client->mqtt_state.in_buffer_read_len, read_poll_timeout_ms);
read_len = esp_transport_read(t, (char *)buf, total_len - client->mqtt_state.inbound_message.read_len, read_poll_timeout_ms);
ESP_LOGD(TAG, "%s: read_len=%d", __func__, read_len);
if (read_len <= 0) {
return esp_mqtt_handle_transport_read_error(read_len, client, true);
}
client->mqtt_state.in_buffer_read_len += read_len;
client->mqtt_state.inbound_message.read_len += read_len;
if (client->mqtt_state.in_buffer_read_len < total_len) {
if (client->mqtt_state.inbound_message.read_len < total_len) {
ESP_LOGD(TAG,
"%s: transport_read(): message reading left in progress :: total message length: %d (already read: %"NEWLIB_NANO_COMPAT_FORMAT")",
__func__, total_len, NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.in_buffer_read_len));
__func__, total_len, NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.read_len));
return 0;
}
}
ESP_LOGV(TAG, "%s: transport_read():%"NEWLIB_NANO_COMPAT_FORMAT" %"NEWLIB_NANO_COMPAT_FORMAT, __func__,
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.in_buffer_read_len),
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.message_length));
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.read_len),
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.length));
return 1;
err:
esp_mqtt_client_dispatch_transport_error(client);
@@ -1566,7 +1568,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
{
uint8_t msg_type = 0, msg_qos = 0;
uint16_t msg_id = 0;
size_t previous_in_buffer_read_len = client->mqtt_state.in_buffer_read_len;
size_t previous_in_buffer_read_len = client->mqtt_state.inbound_message.read_len;
/* non-blocking receive in order not to block other tasks */
int recv = mqtt_message_receive(client, 0);
@@ -1575,7 +1577,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
}
if (recv == -1) { // Mid-message timeout
if (previous_in_buffer_read_len == client->mqtt_state.in_buffer_read_len) {
if (previous_in_buffer_read_len == client->mqtt_state.inbound_message.read_len) {
// Report error only if didn't receive anything since previous iteration
ESP_LOGE(TAG, "%s: Network timeout while reading MQTT message", __func__);
return ESP_FAIL;
@@ -1589,17 +1591,17 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
return ESP_FAIL;
}
int read_len = client->mqtt_state.message_length;
int read_len = client->mqtt_state.inbound_message.length;
// If the message was valid, get the type, quality of service and id of the message
msg_type = mqtt_get_type(client->mqtt_state.in_buffer);
msg_qos = mqtt_get_qos(client->mqtt_state.in_buffer);
msg_type = mqtt_get_type(client->mqtt_state.inbound_message.data);
msg_qos = mqtt_get_qos(client->mqtt_state.inbound_message.data);
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
msg_id = mqtt5_get_id(client->mqtt_state.in_buffer, read_len);
msg_id = mqtt5_get_id(client->mqtt_state.inbound_message.data, read_len);
#endif
} else {
msg_id = mqtt_get_id(client->mqtt_state.in_buffer, read_len);
msg_id = mqtt_get_id(client->mqtt_state.inbound_message.data, read_len);
}
ESP_LOGD(TAG, "msg_type=%d, msg_id=%d", msg_type, msg_id);
@@ -1612,8 +1614,8 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
#endif
ESP_LOGD(TAG,
"deliver_suback, message_length_read=%"NEWLIB_NANO_COMPAT_FORMAT", message_length=%"NEWLIB_NANO_COMPAT_FORMAT,
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.in_buffer_read_len),
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.message_length));
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.read_len),
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.length));
if (deliver_suback(client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to deliver suback message id=%d", msg_id);
@@ -1638,8 +1640,8 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
case MQTT_MSG_TYPE_PUBLISH:
ESP_LOGD(TAG,
"deliver_publish, message_length_read=%"NEWLIB_NANO_COMPAT_FORMAT", message_length=%"NEWLIB_NANO_COMPAT_FORMAT,
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.in_buffer_read_len),
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.message_length));
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.read_len),
NEWLIB_NANO_COMPAT_CAST(client->mqtt_state.inbound_message.length));
if (deliver_publish(client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to deliver publish message id=%d", msg_id);
@@ -1704,8 +1706,8 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBREC return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_read_len));
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBREC return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data,
client->mqtt_state.inbound_message.read_len));
mqtt5_msg_pubrel(&client->mqtt_state.connection.outbound_message, msg_id);
#endif
} else {
@@ -1726,8 +1728,8 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
if (client->mqtt_state.connection.information.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBREL return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_read_len));
ESP_LOGI(TAG, "MQTT_MSG_TYPE_PUBREL return code is %d", mqtt5_msg_get_reason_code(client->mqtt_state.inbound_message.data,
client->mqtt_state.inbound_message.read_len));
mqtt5_msg_pubcomp(&client->mqtt_state.connection.outbound_message, msg_id);
#endif
} else {
@@ -1795,7 +1797,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
break;
}
client->mqtt_state.in_buffer_read_len = 0;
client->mqtt_state.inbound_message.read_len = 0;
return ESP_OK;
}
@@ -2001,7 +2003,7 @@ static void esp_mqtt_task(void *pv)
client->event.event_id = MQTT_EVENT_CONNECTED;
if (client->mqtt_state.connection.information.protocol_ver != MQTT_PROTOCOL_V_5) {
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.in_buffer);
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.inbound_message.data);
}
client->state = MQTT_STATE_CONNECTED;