Skip to content

Commit

Permalink
Adds error code to MQTT_EVENT_SUBSCRIBED in case of failure
Browse files Browse the repository at this point in the history 
- Payload data parsed for error codes. Check only for first filter
  because our subscribe function only handle one topic filter in each
  call.

- Closes #233
  • Loading branch information
@euripedesrocha
euripedesrocha committed Oct 20, 2022
1 parent 2b52b50 commit 9af5c26
Show file tree
Hide file tree
Showing 2 changed files with 49 additions and 37 deletions.
7 changes: 4 additions & 3 deletions include/mqtt_client.h
View file
Expand Up @@ -46,12 +46,12 @@ typedef enum esp_mqtt_event_id_t {
MQTT_EVENT_DISCONNECTED, /*!< disconnected event */
MQTT_EVENT_SUBSCRIBED, /*!< subscribed event, additional context:
- msg_id message id
- data pointer to the received
data
- error_handle `error_type` in case subscribing failed
- data pointer to broker response, check for errors.
- data_len length of the data for this
event
*/
MQTT_EVENT_UNSUBSCRIBED, /*!< unsubscribed event */
MQTT_EVENT_UNSUBSCRIBED, /*!< unsubscribed event, additional context: msg_id */
MQTT_EVENT_PUBLISHED, /*!< published event, additional context: msg_id */
MQTT_EVENT_DATA, /*!< data event, additional context:
- msg_id message id
Expand Down Expand Up @@ -112,6 +112,7 @@ typedef enum esp_mqtt_error_type_t {
MQTT_ERROR_TYPE_NONE = 0,
MQTT_ERROR_TYPE_TCP_TRANSPORT,
MQTT_ERROR_TYPE_CONNECTION_REFUSED,
MQTT_ERROR_TYPE_SUBSCRIBE_FAILED
} esp_mqtt_error_type_t;

/**
Expand Down
79 changes: 45 additions & 34 deletions mqtt_client.c
View file
@@ -1,5 +1,7 @@
#include "esp_log.h"
#include "mqtt_client.h"
#include "mqtt_client_priv.h"
#include "esp_log.h"
#include <stdint.h>

_Static_assert(sizeof(uint64_t) == sizeof(outbox_tick_t), "mqtt-client tick type size different from outbox tick type");
#ifdef ESP_EVENT_ANY_ID
Expand Down Expand Up @@ -588,9 +590,10 @@ void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
client->config = NULL;
}

static inline bool has_timed_out(uint64_t last_tick, uint64_t timeout) {
uint64_t next = last_tick + timeout;
return (int64_t)(next - platform_tick_get_ms()) <= 0;
static inline bool has_timed_out(uint64_t last_tick, uint64_t timeout)
{
uint64_t next = last_tick + timeout;
return (int64_t)(next - platform_tick_get_ms()) <= 0;
}

static esp_err_t process_keepalive(esp_mqtt_client_handle_t client)
Expand All @@ -608,7 +611,7 @@ static esp_err_t process_keepalive(esp_mqtt_client_handle_t client)
return ESP_OK;
}

if (has_timed_out(client->keepalive_tick, keepalive_ms/2)) {
if (has_timed_out(client->keepalive_tick, keepalive_ms / 2)) {
if (esp_mqtt_client_ping(client) == ESP_FAIL) {
ESP_LOGE(TAG, "Can't send ping, disconnected");
esp_mqtt_abort_connection(client);
Expand Down Expand Up @@ -649,11 +652,11 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_connect(&client->mqtt_state.mqtt_connection,
&client->connect_info, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->will_property_info);
&client->connect_info, &client->mqtt5_config->connect_property_info, &client->mqtt5_config->will_property_info);
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection,
&client->connect_info);
&client->connect_info);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Connect message cannot be created");
Expand All @@ -664,11 +667,11 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.pending_msg_id = mqtt5_get_id(client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
client->mqtt_state.outbound_message->length);
#endif
} else {
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
client->mqtt_state.outbound_message->length);
}
ESP_LOGD(TAG, "Sending MQTT CONNECT message, type: %d, id: %04X",
client->mqtt_state.pending_msg_type,
Expand Down Expand Up @@ -768,11 +771,11 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
{
esp_mqtt_client_handle_t client = heap_caps_calloc(1, sizeof(struct esp_mqtt_client),
#if MQTT_EVENT_QUEUE_SIZE > 1
// if supporting multiple queued events, we keep track of them
// using atomic variable, so need to make sure it won't get allocated in PSRAM
MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT);
// if supporting multiple queued events, we keep track of them
// using atomic variable, so need to make sure it won't get allocated in PSRAM
MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT);
#else
MALLOC_CAP_DEFAULT);
MALLOC_CAP_DEFAULT);
#endif
ESP_MEM_CHECK(TAG, client, return NULL);
if (!create_client_data(client)) {
Expand Down Expand Up @@ -1049,8 +1052,8 @@ static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
msg_topic_len = 0;
msg_data_offset += msg_data_len;
int ret = esp_transport_read(client->transport, (char *)client->mqtt_state.in_buffer,
msg_total_len - msg_read_len > buf_len ? buf_len : msg_total_len - msg_read_len,
client->config->network_timeout_ms);
msg_total_len - msg_read_len > buf_len ? buf_len : msg_total_len - msg_read_len,
client->config->network_timeout_ms);
if (ret <= 0) {
return esp_mqtt_handle_transport_read_error(ret, client) == 0 ? ESP_OK : ESP_FAIL;
}
Expand Down Expand Up @@ -1079,7 +1082,15 @@ static esp_err_t deliver_suback(esp_mqtt_client_handle_t client)
ESP_LOGE(TAG, "Failed to acquire suback data");
return ESP_FAIL;
}
client->event.error_handle->esp_tls_stack_err = 0;
client->event.error_handle->esp_tls_last_esp_err = 0;
client->event.error_handle->esp_tls_cert_verify_flags = 0;
client->event.error_handle->error_type = MQTT_ERROR_TYPE_NONE;
client->event.error_handle->connect_return_code = MQTT_CONNECTION_ACCEPTED;
// post data event
if ((uint8_t)*msg_data >= 0x80) {
client->event.error_handle->error_type = MQTT_ERROR_TYPE_SUBSCRIBE_FAILED;
}
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
client->event.event_id = MQTT_EVENT_SUBSCRIBED;
Expand Down Expand Up @@ -1425,7 +1436,7 @@ static esp_err_t mqtt_resend_queued(esp_mqtt_client_handle_t client, outbox_item
// set duplicate flag for QoS-1 and QoS-2 messages
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_publish_qos > 0 && (outbox_item_get_pending(item) == TRANSMITTED)) {
mqtt_set_dup(client->mqtt_state.outbound_message->data);
ESP_LOGD(TAG,"Sending Duplicated QoS%d message with id=%d", client->mqtt_state.pending_publish_qos, client->mqtt_state.pending_msg_id);
ESP_LOGD(TAG, "Sending Duplicated QoS%d message with id=%d", client->mqtt_state.pending_publish_qos, client->mqtt_state.pending_msg_id);
}

// try to resend the data
Expand Down Expand Up @@ -1477,17 +1488,17 @@ static inline int max_poll_timeout(esp_mqtt_client_handle_t client, int max_time
{
return
#if MQTT_EVENT_QUEUE_SIZE > 1
atomic_load(&client->queued_events) > 0 ? 10: max_timeout;
atomic_load(&client->queued_events) > 0 ? 10 : max_timeout;
#else
max_timeout;
max_timeout;
#endif
}

static inline void run_event_loop(esp_mqtt_client_handle_t client)
{
#if MQTT_EVENT_QUEUE_SIZE > 1
if (atomic_load(&client->queued_events) > 0) {
atomic_fetch_sub(&client->queued_events, 1);
atomic_fetch_sub(&client->queued_events, 1);
#else
{
#endif
Expand Down Expand Up @@ -1599,7 +1610,7 @@ static void esp_mqtt_task(void *pv)
}

if (client->config->refresh_connection_after_ms &&
has_timed_out(client->refresh_connection_tick, client->config->refresh_connection_after_ms)) {
has_timed_out(client->refresh_connection_tick, client->config->refresh_connection_after_ms)) {
ESP_LOGD(TAG, "Refreshing the connection...");
esp_mqtt_abort_connection(client);
client->state = MQTT_STATE_INIT;
Expand Down Expand Up @@ -1805,16 +1816,16 @@ int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic
return -1;
}
client->mqtt_state.outbound_message = mqtt5_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
&client->mqtt_state.pending_msg_id, client->mqtt5_config->subscribe_property_info);
topic, qos,
&client->mqtt_state.pending_msg_id, client->mqtt5_config->subscribe_property_info);
if (client->mqtt_state.outbound_message->length) {
client->mqtt5_config->subscribe_property_info = NULL;
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
&client->mqtt_state.pending_msg_id);
topic, qos,
&client->mqtt_state.pending_msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
ESP_LOGE(TAG, "Subscribe message cannot be created");
Expand Down Expand Up @@ -1857,16 +1868,16 @@ int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *top
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
topic,
&client->mqtt_state.pending_msg_id, client->mqtt5_config->unsubscribe_property_info);
topic,
&client->mqtt_state.pending_msg_id, client->mqtt5_config->unsubscribe_property_info);
if (client->mqtt_state.outbound_message->length) {
client->mqtt5_config->unsubscribe_property_info = NULL;
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
topic,
&client->mqtt_state.pending_msg_id);
topic,
&client->mqtt_state.pending_msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
MQTT_API_UNLOCK(client);
Expand Down Expand Up @@ -1901,18 +1912,18 @@ static inline int mqtt_client_enqueue_priv(esp_mqtt_client_handle_t client, cons
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
client->mqtt_state.outbound_message = mqtt5_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, len,
qos, retain,
&pending_msg_id, client->mqtt5_config->publish_property_info, client->mqtt5_config->server_resp_property_info.response_info);
topic, data, len,
qos, retain,
&pending_msg_id, client->mqtt5_config->publish_property_info, client->mqtt5_config->server_resp_property_info.response_info);
if (client->mqtt_state.outbound_message->length) {
client->mqtt5_config->publish_property_info = NULL;
}
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, len,
qos, retain,
&pending_msg_id);
topic, data, len,
qos, retain,
&pending_msg_id);
}

if (client->mqtt_state.outbound_message->length == 0) {
Expand Down

0 comments on commit 9af5c26

Please sign in to comment.