configure MQTT callback in non blocking mode for esp32 using esp-idf 4.2

[GirishHolla]

I'm using mqtt callback function 'esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event)' in esp-idf 4.2 but this function is blocking 'while(1)' loop in main function. Kindly let me know for the solution to run event based mqtt callback function in non blocking mode. Thank you in advance.

[ESP_Sprite]

...as far as I understand, that is a function you provide yourself, so if that function blocks, I'd look at what you wrote there.

[GirishHolla]

[/*********call back function ********************************/
static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {

const char *TAG = "MQTT_CB";
esp_mqtt_client_handle_t client = event->client;
int msg_id;
// your_context_t *context = event->context;

switch (event->event_id) {
case MQTT_EVENT_CONNECTED:
flag.mqtt_evt_cnted = 1;
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
msg_id = esp_mqtt_client_subscribe(client, "esp32", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
break;

case MQTT_EVENT_DISCONNECTED:
flag.mqtt_evt_cnted = 0;
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
ESP_LOGI(TAG, "retried to connect.....");
esp_wifi_connect();
break;

case MQTT_EVENT_SUBSCRIBED:

ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
msg_id = esp_mqtt_client_publish(client, "esp32", "data",
4, 1, 0);
ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_PUBLISHED:
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_DATA:

ESP_LOGI(TAG, "MQTT_EVENT_DATA");
printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
printf("DATA=%.*s\r\n", event->data_len, event->data);

break;
case MQTT_EVENT_ERROR:
ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
break;
default:
ESP_LOGI(TAG, "Other event id:%d", event->event_id);
break;
}
return ESP_OK;
}





/**********creating task to upload @ particular event******************/
void publish_task(void *pvParameters) {

char TAG[] = "PUB_TASK";
int msg_id;
uint8_t one_time_flag = 0;
esp_mqtt_client_config_t mqtt_cfg = { .uri =
"mqtt://mqtt.eclipseprojects.io:1883",
};
esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);

if (one_time_flag == 1) {
one_time_flag = 0;
esp_mqtt_client_register_event(client, ESP_EVENT_ANY_ID,
mqtt_event_handler, client);
esp_mqtt_client_start(client);
}

while (1) { //if while is removed data is not uploded @ particular interval
if (publish data at perticular event) {
msg_id = esp_mqtt_client_publish(client, "esp32", data_buf, 0, 1,0); // used task to loop publish the data from BT
ESP_LOGI(TAG, "message ID %d", msg_id);
}

}
}
]

[ESP_Sprite]

As far as I can see, there's nothing in there that is blocking for a non-trivial amount of time (logging can block on the UART output function, but that's not for more than a few ms at max). What exactly is the issue you're running into with this code?

[markkuk]

The posted code never registers (or calls) the event handler, because variable one_time_flag is initialized as zero.

[GirishHolla]

Hi, @markkuk yes normally i tried but loop wise upload to mqtt server is works fine. Problem this publish task not making to enter while (1) loop in app_main for esp32.

[GirishHolla]

I will be explaining again with my issue.

1. Created publish task to upload MQTT bluetooth data with every cycle. Here its results with two error i.e.,
a. when the function called uart_set_pin called (in uart_init function for UART 2 in esp32) this blocking cyclic publish for mqtt. If uart_set_pin is commented then cyclic publish works fine.
b. One more thing publish_task blocking while(1) in the main function (screenshot shared in my previous reply for the same post)(when the line commented as explained as mentioned as above).

/**************Uart init function***********************/
void uart_init(void){

const uart_port_t uart_num = UART_NUM_2;
uart_config_t uart_config = {
.baud_rate = 115200,
.data_bits = UART_DATA_8_BITS,
.parity = UART_PARITY_DISABLE,
.stop_bits = UART_STOP_BITS_1,
.flow_ctrl = UART_HW_FLOWCTRL_DISABLE,
.rx_flow_ctrl_thresh = 122,
};
uart_param_config(uart_num, &uart_config);
uart_driver_install(uart_num, BUF_SIZE * 2, BUF_SIZE * 2, 0, NULL, 0);
// uart_set_pin(uart_num, UART2_Tx_PIN, UART2_Rx_PIN, UART_PIN_NO_CHANGE, UART_PIN_NO_CHANGE);
}

void publish_task(void *pvParameters) {

char TAG[] = "PUB_TASK";
int msg_id;
esp_mqtt_client_config_t mqtt_cfg = { .uri =
"mqtt://mqtt.eclipseprojects.io",
.port = 1883,
};
esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);

esp_mqtt_client_register_event(client, ESP_EVENT_ANY_ID,
mqtt_event_handler, client);
esp_mqtt_client_start(client);

while (1) {
if (flag.upload_bcn_data == 1) {
flag.upload_bcn_data = 0;
msg_id = esp_mqtt_client_publish(client, "esp32", data_buf, 0, 1,
0);
ESP_LOGI(TAG, "message ID %d", msg_id);
}

}
}


/*********call abck for mqtt*************************/
static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event) {

const char *TAG = "MQTT_CB";
esp_mqtt_client_handle_t client = event->client;
int msg_id;
// your_context_t *context = event->context;

switch (event->event_id) {
case MQTT_EVENT_CONNECTED:
flag.mqtt_evt_cnted = 1;
ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
msg_id = esp_mqtt_client_subscribe(client, "esp32", 1);
ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);
break;

case MQTT_EVENT_DISCONNECTED:
flag.mqtt_evt_cnted = 0;
ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
ESP_LOGI(TAG, "retried to connect.....");
esp_wifi_connect();
break;

case MQTT_EVENT_SUBSCRIBED:
//esp_task_wdt_reset(); /*feed watch dog*/

ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
/* msg_id = esp_mqtt_client_publish(client, "esp32", "data",
4, 1, 0);
ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);*/
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_PUBLISHED:
// esp_task_wdt_reset(); /*feed watch dog*/
ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
break;
case MQTT_EVENT_DATA:

//esp_task_wdt_reset(); /*feed watch dog*/
ESP_LOGI(TAG, "MQTT_EVENT_DATA");
printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
printf("DATA=%.*s\r\n", event->data_len, event->data);

break;
case MQTT_EVENT_ERROR:
ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
break;
default:
ESP_LOGI(TAG, "Other event id:%d", event->event_id);
break;
}
return ESP_OK;
}

[ESP_Sprite]

What is the value of UART2_Tx_PIN, UART2_Rx_PIN, and what hardware are you using?

[GirishHolla]

@ ESP_Sprite Thank you for your response, actully using GPIO 17 & 16 for UART Tx and UART Rx respectively for ESP-WROOM-32 module. CTS and RTS disabled with -1.

[GirishHolla]

Problem regarding Uart_set_pin resolved. But still, MQTT call back is blocking the main while loop(as shared screenshot previously) can anyone have a solution kindly let us know thank you in advance.