espressif/esp-mqtt
1
0
Fork 1
mirror of https://github.com/espressif/esp-mqtt.git synced 2025-07-31 11:18:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'feature/multiple_subscribe' into 'master'

Browse Source 
Feature:  Enable SUBSCRIBE to multiple topics

See merge request espressif/esp-mqtt!156
This commit is contained in:
Rocha Euripedes
2023-02-08 15:35:31 +08:00
parent 5adbe11aaf 32102558d3
commit 5a156f56b4
6 changed files with 251 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
include/mqtt_client.h
View File

@ -32,7 +32,7 @@ typedef struct esp_mqtt_client *esp_mqtt_client_handle_t;
* @brief *MQTT* event types.
*
* User event handler receives context data in `esp_mqtt_event_t` structure with
* - `client` - *MQTT* client handle
* - client - *MQTT* client handle
* - various other data depending on event type
*
*/
@ -349,6 +349,14 @@ typedef struct esp_mqtt_client_config_t {
} buffer; /*!< Buffer size configuration.*/
} esp_mqtt_client_config_t;
/**
* Topic definition struct
*/
typedef struct topic_t {
const char *filter; /*!< Topic filter to subscribe */
int qos; /*!< Max QoS level of the subscription */
} esp_mqtt_topic_t;
/**
* @brief Creates *MQTT* client handle based on the configuration
*
@ -417,6 +425,26 @@ esp_err_t esp_mqtt_client_disconnect(esp_mqtt_client_handle_t client);
*/
esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client);
/**
* @brief Convenience macro to select subscribe function to use.
*
* Notes:
* - Usage of `esp_mqtt_client_subscribe_single` is the same as previous
* esp_mqtt_client_subscribe, refer to it for details.
*
* @param client_handle *MQTT* client handle
* @param topic_type Needs to be char* for single subscription or `esp_mqtt_topic_t` for multiple topics
* @param qos_or_size It's either a qos when subscribing to a single topic or the size of the subscription array when subscribing to multiple topics.
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
#define esp_mqtt_client_subscribe(client_handle, topic_type, qos_or_size) _Generic((topic_type), \
char *: esp_mqtt_client_subscribe_single, \
esp_mqtt_topic_t*: esp_mqtt_client_subscribe_multiple \
)(client_handle, topic_type, qos_or_size)
/**
* @brief Subscribe the client to defined topic with defined qos
*
@ -426,23 +454,44 @@ esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client);
* from a *MQTT* event callback i.e. internal *MQTT* task
* (API is protected by internal mutex, so it might block
* if a longer data receive operation is in progress.
* - `esp_mqtt_client_subscribe` could be used to call this function.
*
* @param client *MQTT* client handle
* @param topic
* @param qos // TODO describe parameters
* @param topic topic filter to subscribe
* @param qos Max qos level of the subscription
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client,
int esp_mqtt_client_subscribe_single(esp_mqtt_client_handle_t client,
const char *topic, int qos);
/**
* @brief Subscribe the client to a list of defined topics with defined qos
*
* Notes:
* - Client must be connected to send subscribe message
* - This API is could be executed from a user task or
* from a *MQTT* event callback i.e. internal *MQTT* task
* (API is protected by internal mutex, so it might block
* if a longer data receive operation is in progress.
* - `esp_mqtt_client_subscribe` could be used to call this function.
*
* @param client *MQTT* client handle
* @param topic_list List of topics to subscribe
* @param size size of topic_list
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
int esp_mqtt_client_subscribe_multiple(esp_mqtt_client_handle_t client,
const esp_mqtt_topic_t *topic_list, int size);
/**
* @brief Unsubscribe the client from defined topic
*
* Notes:
* - Client must be connected to send unsubscribe message
* - It is thread safe, please refer to `esp_mqtt_client_subscribe` for details
* - It is thread safe, please refer to `esp_mqtt_client_subscribe_single` for details
*
* @param client *MQTT* client handle
* @param topic

2
lib/include/mqtt5_msg.h
View File

@ -126,7 +126,7 @@ mqtt_message_t *mqtt5_msg_connect(mqtt_connection_t *connection, mqtt_connect_in
mqtt_message_t *mqtt5_msg_publish(mqtt_connection_t *connection, const char *topic, const char *data, int data_length, int qos, int retain, uint16_t *message_id, const esp_mqtt5_publish_property_config_t *property, const char *resp_info);
esp_err_t mqtt5_msg_parse_connack_property(uint8_t *buffer, size_t buffer_len, mqtt_connect_info_t *connection_info, esp_mqtt5_connection_property_storage_t *connection_property, esp_mqtt5_connection_server_resp_property_t *resp_property, int *reason_code, uint8_t *ack_flag, mqtt5_user_property_handle_t *user_property);
int mqtt5_msg_get_reason_code(uint8_t *buffer, size_t length);
mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const char *topic, int qos, uint16_t *message_id, const esp_mqtt5_subscribe_property_config_t *property);
mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const esp_mqtt_topic_t *topic, int size, uint16_t *message_id, const esp_mqtt5_subscribe_property_config_t *property);
mqtt_message_t *mqtt5_msg_unsubscribe(mqtt_connection_t *connection, const char *topic, uint16_t *message_id, const esp_mqtt5_unsubscribe_property_config_t *property);
mqtt_message_t *mqtt5_msg_disconnect(mqtt_connection_t *connection, esp_mqtt5_disconnect_property_config_t *disconnect_property_info);
mqtt_message_t *mqtt5_msg_pubcomp(mqtt_connection_t *connection, uint16_t message_id);

2
lib/include/mqtt_msg.h
View File

@ -138,7 +138,7 @@ mqtt_message_t *mqtt_msg_puback(mqtt_connection_t *connection, uint16_t message_
mqtt_message_t *mqtt_msg_pubrec(mqtt_connection_t *connection, uint16_t message_id);
mqtt_message_t *mqtt_msg_pubrel(mqtt_connection_t *connection, uint16_t message_id);
mqtt_message_t *mqtt_msg_pubcomp(mqtt_connection_t *connection, uint16_t message_id);
mqtt_message_t *mqtt_msg_subscribe(mqtt_connection_t *connection, const char *topic, int qos, uint16_t *message_id);
mqtt_message_t *mqtt_msg_subscribe(mqtt_connection_t *connection, const esp_mqtt_topic_t topic_list[], int size, uint16_t *message_id) __attribute__((nonnull));
mqtt_message_t *mqtt_msg_unsubscribe(mqtt_connection_t *connection, const char *topic, uint16_t *message_id);
mqtt_message_t *mqtt_msg_pingreq(mqtt_connection_t *connection);
mqtt_message_t *mqtt_msg_pingresp(mqtt_connection_t *connection);

27
lib/mqtt5_msg.c
View File

@ -1,5 +1,6 @@
#include <string.h>
#include "mqtt5_msg.h"
#include "mqtt_client.h"
#include "mqtt_config.h"
#include "platform.h"
#include "esp_log.h"
@ -764,7 +765,7 @@ mqtt_message_t *mqtt5_msg_publish(mqtt_connection_t *connection, const char *top
}
snprintf(response_topic, response_topic_size, "%s/%s", property->response_topic, resp_info);
if (append_property(connection, MQTT5_PROPERTY_RESPONSE_TOPIC, 2, response_topic, response_topic_size) == -1) {
ESP_LOGE(TAG,"%s(%d) fail",__FUNCTION__, __LINE__);
ESP_LOGE(TAG, "%s(%d) fail", __FUNCTION__, __LINE__);
free(response_topic);
return fail_message(connection);
}
@ -849,14 +850,10 @@ int mqtt5_msg_get_reason_code(uint8_t *buffer, size_t length)
return -1;
}
mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const char *topic, int qos, uint16_t *message_id, const esp_mqtt5_subscribe_property_config_t *property)
mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const esp_mqtt_topic_t *topic_list, int size, uint16_t *message_id, const esp_mqtt5_subscribe_property_config_t *property)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
}
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
@ -877,22 +874,27 @@ mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const char *t
}
}
APPEND_CHECK(update_property_len_value(connection, connection->message.length - properties_offset - 1, properties_offset), fail_message(connection));
for (int topic_number = 0; topic_number < size; ++topic_number) {
if (topic_list[topic_number].filter[0] == '\0') {
return fail_message(connection);
}
if (property && property->is_share_subscribe) {
uint16_t shared_topic_size = strlen(topic) + strlen(MQTT5_SHARED_SUB) + strlen(property->share_name);
uint16_t shared_topic_size = strlen(topic_list[topic_number].filter) + strlen(MQTT5_SHARED_SUB) + strlen(property->share_name);
char *shared_topic = calloc(1, shared_topic_size);
if (!shared_topic) {
ESP_LOGE(TAG, "Failed to calloc %d memory", shared_topic_size);
fail_message(connection);
}
snprintf(shared_topic, shared_topic_size, MQTT5_SHARED_SUB, property->share_name, topic);
snprintf(shared_topic, shared_topic_size, MQTT5_SHARED_SUB, property->share_name, topic_list[topic_number].filter);
if (append_property(connection, 0, 2, shared_topic, strlen(shared_topic)) == -1) {
ESP_LOGE(TAG,"%s(%d) fail",__FUNCTION__, __LINE__);
ESP_LOGE(TAG, "%s(%d) fail", __FUNCTION__, __LINE__);
free(shared_topic);
return fail_message(connection);
}
free(shared_topic);
} else {
APPEND_CHECK(append_property(connection, 0, 2, topic, strlen(topic)), fail_message(connection));
APPEND_CHECK(append_property(connection, 0, 2, topic_list[topic_number].filter, strlen(topic_list[topic_number].filter)), fail_message(connection));
}
if (connection->message.length + 1 > connection->buffer_length) {
@ -910,8 +912,9 @@ mqtt_message_t *mqtt5_msg_subscribe(mqtt_connection_t *connection, const char *t
connection->buffer[connection->message.length] |= (property->retain_as_published_flag << 3);
}
}
connection->buffer[connection->message.length] |= (qos & 3);
connection->buffer[connection->message.length] |= (topic_list[topic_number].qos & 3);
connection->message.length ++;
}
return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0);
}
@ -975,7 +978,7 @@ mqtt_message_t *mqtt5_msg_unsubscribe(mqtt_connection_t *connection, const char
}
snprintf(shared_topic, shared_topic_size, MQTT5_SHARED_SUB, property->share_name, topic);
if (append_property(connection, 0, 2, shared_topic, strlen(shared_topic)) == -1) {
ESP_LOGE(TAG,"%s(%d) fail",__FUNCTION__, __LINE__);
ESP_LOGE(TAG, "%s(%d) fail", __FUNCTION__, __LINE__);
free(shared_topic);
return fail_message(connection);
}

18
lib/mqtt_msg.c
View File

@ -29,6 +29,7 @@
*
*/
#include <string.h>
#include "mqtt_client.h"
#include "mqtt_msg.h"
#include "mqtt_config.h"
#include "platform.h"
@ -518,26 +519,29 @@ mqtt_message_t *mqtt_msg_pubcomp(mqtt_connection_t *connection, uint16_t message
return fini_message(connection, MQTT_MSG_TYPE_PUBCOMP, 0, 0, 0);
}
mqtt_message_t *mqtt_msg_subscribe(mqtt_connection_t *connection, const char *topic, int qos, uint16_t *message_id)
mqtt_message_t *mqtt_msg_subscribe(mqtt_connection_t *connection, const esp_mqtt_topic_t topic_list[], int size, uint16_t *message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
}
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
if (append_string(connection, topic, strlen(topic)) < 0) {
for (int topic_number = 0; topic_number < size; ++topic_number) {
if (topic_list[topic_number].filter[0] == '\0') {
return fail_message(connection);
}
if (append_string(connection, topic_list[topic_number].filter, strlen(topic_list[topic_number].filter)) < 0) {
return fail_message(connection);
}
if (connection->message.length + 1 > connection->buffer_length) {
return fail_message(connection);
}
connection->buffer[connection->message.length++] = qos;
connection->buffer[connection->message.length] = topic_list[topic_number].qos;
connection->message.length ++;
}
return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0);
}

44
mqtt_client.c
View File

@ -1101,8 +1101,11 @@ static esp_err_t deliver_suback(esp_mqtt_client_handle_t client)
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) {
for (int topic = 0; topic < msg_data_len; ++topic) {
if ((uint8_t)msg_data[topic] == 0x80) {
client->event.error_handle->error_type = MQTT_ERROR_TYPE_SUBSCRIBE_FAILED;
break;
}
}
client->event.data_len = msg_data_len;
client->event.total_data_len = msg_data_len;
@ -1114,7 +1117,9 @@ static esp_err_t deliver_suback(esp_mqtt_client_handle_t client)
return ESP_OK;
}
static bool is_valid_mqtt_msg(esp_mqtt_client_handle_t client, int msg_type, int msg_id)
// Deletes the initial message in MQTT communication protocol
// Return false when message is not found, making the received counterpart invalid.
static bool remove_initiator_message(esp_mqtt_client_handle_t client, int msg_type, int msg_id)
{
ESP_LOGD(TAG, "pending_id=%d, pending_msg_count = %d", client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_count);
if (client->mqtt_state.pending_msg_count == 0) {
@ -1311,7 +1316,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
switch (msg_type) {
case MQTT_MSG_TYPE_SUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_SUBSCRIBE, msg_id)) {
if (remove_initiator_message(client, MQTT_MSG_TYPE_SUBSCRIBE, msg_id)) {
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_suback(client);
#endif
@ -1323,7 +1328,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
}
break;
case MQTT_MSG_TYPE_UNSUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_UNSUBSCRIBE, msg_id)) {
if (remove_initiator_message(client, MQTT_MSG_TYPE_UNSUBSCRIBE, msg_id)) {
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_unsuback(client);
#endif
@ -1375,7 +1380,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
esp_mqtt5_decrement_packet_counter(client);
}
#endif
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
if (remove_initiator_message(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish");
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_puback(client);
@ -1426,7 +1431,7 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
esp_mqtt5_decrement_packet_counter(client);
}
#endif
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
if (remove_initiator_message(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
#ifdef MQTT_PROTOCOL_5
esp_mqtt5_parse_pubcomp(client);
@ -1820,7 +1825,8 @@ static esp_err_t esp_mqtt_client_ping(esp_mqtt_client_handle_t client)
return ESP_OK;
}
int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos)
int esp_mqtt_client_subscribe_multiple(esp_mqtt_client_handle_t client,
const esp_mqtt_topic_t *topic_list, int size)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
@ -1834,13 +1840,19 @@ int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic
}
if (client->connect_info.protocol_ver == MQTT_PROTOCOL_V_5) {
#ifdef MQTT_PROTOCOL_5
if (esp_mqtt5_client_subscribe_check(client, qos) != ESP_OK) {
ESP_LOGI(TAG, "MQTT5 subscribe check fail");
int max_qos = topic_list[0].qos;
for (int topic_number = 0; topic_number < size; ++topic_number) {
if (topic_list[topic_number].qos > max_qos) {
max_qos = topic_list[topic_number].qos;
}
}
if (esp_mqtt5_client_subscribe_check(client, max_qos) != ESP_OK) {
ESP_LOGI(TAG, "MQTT5 subscribe check fail: QoS %d not accepted by broker ", max_qos);
MQTT_API_UNLOCK(client);
return -1;
}
client->mqtt_state.outbound_message = mqtt5_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
topic_list, size,
&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;
@ -1848,7 +1860,7 @@ int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic
#endif
} else {
client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
topic, qos,
topic_list, size,
&client->mqtt_state.pending_msg_id);
}
if (client->mqtt_state.outbound_message->length == 0) {
@ -1867,14 +1879,20 @@ int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic
outbox_set_pending(client->outbox, client->mqtt_state.pending_msg_id, TRANSMITTED);// handle error
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error to subscribe topic=%s, qos=%d", topic, qos);
ESP_LOGE(TAG, "Error to send subscribe message, first topic: %s, qos: %d", topic_list[0].filter, topic_list[0].qos);
MQTT_API_UNLOCK(client);
return -1;
}
ESP_LOGD(TAG, "Sent subscribe topic=%s, id: %d, type=%d successful", topic, client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_type);
ESP_LOGD(TAG, "Sent subscribe, first topic=%s, id: %d", topic_list[0].filter, client->mqtt_state.pending_msg_id);
MQTT_API_UNLOCK(client);
return client->mqtt_state.pending_msg_id;
}
int esp_mqtt_client_subscribe_single(esp_mqtt_client_handle_t client, const char *topic, int qos)
{
esp_mqtt_topic_t user_topic = {.filter = topic, .qos = qos};
return esp_mqtt_client_subscribe_multiple(client, &user_topic, 1);
}
int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic)