feedc0de/esp-mqtt
1
0
Fork 0
forked from espressif/esp-mqtt
Code Pull Requests Activity

Merge branch 'feature/reliability_improvemnts' into 'idf'

Browse Source 
MQTT reliability improvemnts

See merge request idf/esp-mqtt!30
This commit is contained in:
David Čermák
2019-05-16 03:00:07 +08:00
parent c4fdd7759c 18b6f2c582
commit 6f1fc3785a
11 changed files with 880 additions and 545 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitlab-ci.yml
View File

@ -79,7 +79,7 @@ clang_tidy_check:
- cd $IDF_PATH/components/mqtt/esp-mqtt
- rm -rf .git
- cp -r $CI_PROJECT_DIR/.git .
- git checkout $CI_COMMIT_SHA
- git reset --hard $CI_COMMIT_SHA
- mv $CI_PROJECT_DIR/tidybuild $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild
- cd $IDF_PATH/examples/protocols/mqtt/tcp/tidybuild
- git clone "${CI_REPORT_TOOLS}"

8
.travis.yml
View File

@ -16,6 +16,8 @@ before_install:
# Have to checkout a temp branch for later in tree reference
- git checkout -b temporary_ref_branch
- CI_COMMIT_SHA=$(git rev-parse HEAD)
# Test building with latest (stable == v3.3 for now) IDF
- LATEST_IDF=release/v3.3
install:
# Install ESP32 toochain following steps as desribed
@ -42,13 +44,13 @@ script:
- cd $IDF_PATH
- git checkout v3.1 && git submodule update --init --recursive
- cd $PROJECT_PATH
- ./modify_for_legacy_idf.sh || true
- ./modify_for_legacy_idf.sh ${LATEST_IDF} || true
- cd $PROJECT_PATH/examples/tcp
- make defconfig
- make -j4
# Master build with latest IDF
# Build with latest IDF
- cd $IDF_PATH
- git checkout master && git submodule update --init --recursive
- git checkout ${LATEST_IDF} && git submodule update --init --recursive
- cd $IDF_PATH/components/mqtt/esp-mqtt
- git remote add local $PROJECT_PATH/.git
- git fetch local

127
include/mqtt_client.h Executable file → Normal file
View File

@ -18,7 +18,7 @@
extern "C" {
#endif
typedef struct esp_mqtt_client* esp_mqtt_client_handle_t;
typedef struct esp_mqtt_client *esp_mqtt_client_handle_t;
/**
* @brief MQTT event types.
@ -44,6 +44,10 @@ typedef enum {
- data_len length of the data for this event
- current_data_offset offset of the current data for this event
- total_data_len total length of the data received
Note: Multiple MQTT_EVENT_DATA could be fired for one message, if it is
longer than internal buffer. In that case only first event contains topic
pointer and length, other contain data only with current data length
and current data offset updating.
*/
MQTT_EVENT_BEFORE_CONNECT, /*!< The event occurs before connecting */
} esp_mqtt_event_id_t;
@ -73,7 +77,7 @@ typedef struct {
int session_present; /*!< MQTT session_present flag for connection event */
} esp_mqtt_event_t;
typedef esp_mqtt_event_t* esp_mqtt_event_handle_t;
typedef esp_mqtt_event_t *esp_mqtt_event_handle_t;
typedef esp_err_t (* mqtt_event_callback_t)(esp_mqtt_event_handle_t event);
@ -107,15 +111,130 @@ typedef struct {
int refresh_connection_after_ms; /*!< Refresh connection after this value (in milliseconds) */
} esp_mqtt_client_config_t;
/**
* @brief Creates mqtt client handle based on the configuration
*
* @param config mqtt configuration structure
*
* @return mqtt_client_handle if successfully created, NULL on error
*/
esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config);
/**
* @brief Sets mqtt connection URI. This API is usually used to overrides the URI
* configured in esp_mqtt_client_init
*
* @param client mqtt client hanlde
* @param uri
*
* @return ESP_FAIL if URI parse error, ESP_OK on success
*/
esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *uri);
/**
* @brief Starts mqtt client with already created client handle
*
* @param client mqtt client handle
*
* @return ESP_OK on success
* ESP_ERR_INVALID_ARG on wrong initialization
* ESP_FAIL on other error
*/
esp_err_t esp_mqtt_client_start(esp_mqtt_client_handle_t client);
/**
* @brief This api is typically used to force reconnection upon a specific event
*
* @param client mqtt client handle
*
* @return ESP_OK on success
* ESP_FAIL if client is in invalid state
*/
esp_err_t esp_mqtt_client_reconnect(esp_mqtt_client_handle_t client);
/**
* @brief Stops mqtt client tasks
*
* @param client mqtt client handle
*
* @return ESP_OK on success
* ESP_FAIL if client is in invalid state
*/
esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client);
esp_err_t esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos);
esp_err_t esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic);
/**
* @brief Subscribe the client to defined topic 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.
*
* @param client mqtt client handle
* @param topic
* @param qos
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
int esp_mqtt_client_subscribe(esp_mqtt_client_handle_t client, const char *topic, int qos);
/**
* @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
*
* @param client mqtt client handle
* @param topic
*
* @return message_id of the subscribe message on success
* -1 on failure
*/
int esp_mqtt_client_unsubscribe(esp_mqtt_client_handle_t client, const char *topic);
/**
* @brief Client to send a publish message to the broker
*
* Notes:
* - Client doesn't have to be connected to send publish message
* (although it would drop all qos=0 messages, qos>1 messages would be enqueued)
* - It is thread safe, please refer to `esp_mqtt_client_subscribe` for details
*
* @param client mqtt client handle
* @param topic topic string
* @param data payload string (set to NULL, sending empty payload message)
* @param len data length, if set to 0, length is calculated from payload string
* @param qos qos of publish message
* @param retain ratain flag
*
* @return message_id of the subscribe message on success
* 0 if cannot publish
*/
int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic, const char *data, int len, int qos, int retain);
/**
* @brief Destroys the client handle
*
* @param client mqtt client handle
*
* @return ESP_OK
*/
esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client);
/**
* @brief Set configuration structure, typically used when updating the config (i.e. on "before_connect" event
*
* @param client mqtt client handle
*
* @param config mqtt configuration structure
*
* @return ESP_ERR_NO_MEM if failed to allocate
* ESP_OK on success
*/
esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config);
#ifdef __cplusplus

12
include/mqtt_config.h
View File

@ -67,13 +67,13 @@
#endif
#ifdef CONFIG_MQTT_USE_CORE_0
#define MQTT_TASK_CORE 0
#define MQTT_TASK_CORE 0
#else
#ifdef CONFIG_MQTT_USE_CORE_1
#define MQTT_TASK_CORE 1
#else
#define MQTT_TASK_CORE 0
#endif
#ifdef CONFIG_MQTT_USE_CORE_1
#define MQTT_TASK_CORE 1
#else
#define MQTT_TASK_CORE 0
#endif
#endif

99
lib/include/mqtt_msg.h
View File

@ -40,8 +40,7 @@ extern "C" {
/* Remaining Length */
enum mqtt_message_type
{
enum mqtt_message_type {
MQTT_MSG_TYPE_CONNECT = 1,
MQTT_MSG_TYPE_CONNACK = 2,
MQTT_MSG_TYPE_PUBLISH = 3,
@ -58,8 +57,7 @@ enum mqtt_message_type
MQTT_MSG_TYPE_DISCONNECT = 14
};
enum mqtt_connect_return_code
{
enum mqtt_connect_return_code {
CONNECTION_ACCEPTED = 0,
CONNECTION_REFUSE_PROTOCOL,
CONNECTION_REFUSE_ID_REJECTED,
@ -68,31 +66,28 @@ enum mqtt_connect_return_code
CONNECTION_REFUSE_NOT_AUTHORIZED
};
typedef struct mqtt_message
{
uint8_t* data;
typedef struct mqtt_message {
uint8_t *data;
uint32_t length;
uint32_t fragmented_msg_total_length; /*!< total len of fragmented messages (zero for all other messages) */
uint32_t fragmented_msg_data_offset; /*!< data offset of fragmented messages (zero for all other messages) */
} mqtt_message_t;
typedef struct mqtt_connection
{
typedef struct mqtt_connection {
mqtt_message_t message;
uint16_t message_id;
uint8_t* buffer;
uint8_t *buffer;
uint16_t buffer_length;
} mqtt_connection_t;
typedef struct mqtt_connect_info
{
char* client_id;
char* username;
char* password;
char* will_topic;
char* will_message;
typedef struct mqtt_connect_info {
char *client_id;
char *username;
char *password;
char *will_topic;
char *will_message;
int keepalive;
int will_length;
int will_qos;
@ -102,32 +97,54 @@ typedef struct mqtt_connect_info
} mqtt_connect_info_t;
static inline int mqtt_get_type(uint8_t* buffer) { return (buffer[0] & 0xf0) >> 4; }
static inline int mqtt_get_connect_session_present(uint8_t* buffer) { return buffer[2] & 0x01; }
static inline int mqtt_get_connect_return_code(uint8_t* buffer) { return buffer[3]; }
static inline int mqtt_get_dup(uint8_t* buffer) { return (buffer[0] & 0x08) >> 3; }
static inline void mqtt_set_dup(uint8_t* buffer) { buffer[0] |= 0x08; }
static inline int mqtt_get_qos(uint8_t* buffer) { return (buffer[0] & 0x06) >> 1; }
static inline int mqtt_get_retain(uint8_t* buffer) { return (buffer[0] & 0x01); }
static inline int mqtt_get_type(uint8_t *buffer)
{
return (buffer[0] & 0xf0) >> 4;
}
static inline int mqtt_get_connect_session_present(uint8_t *buffer)
{
return buffer[2] & 0x01;
}
static inline int mqtt_get_connect_return_code(uint8_t *buffer)
{
return buffer[3];
}
static inline int mqtt_get_dup(uint8_t *buffer)
{
return (buffer[0] & 0x08) >> 3;
}
static inline void mqtt_set_dup(uint8_t *buffer)
{
buffer[0] |= 0x08;
}
static inline int mqtt_get_qos(uint8_t *buffer)
{
return (buffer[0] & 0x06) >> 1;
}
static inline int mqtt_get_retain(uint8_t *buffer)
{
return (buffer[0] & 0x01);
}
void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length);
bool mqtt_header_complete(uint8_t* buffer, uint16_t buffer_length);
uint32_t mqtt_get_total_length(uint8_t* buffer, uint16_t length);
const char* mqtt_get_publish_topic(uint8_t* buffer, uint32_t* length);
const char* mqtt_get_publish_data(uint8_t* buffer, uint32_t* length);
uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length);
void mqtt_msg_init(mqtt_connection_t *connection, uint8_t *buffer, uint16_t buffer_length);
bool mqtt_header_complete(uint8_t *buffer, uint16_t buffer_length);
uint32_t mqtt_get_total_length(uint8_t *buffer, uint16_t length, int *fixed_size_len);
char *mqtt_get_publish_topic(uint8_t *buffer, uint32_t *length);
char *mqtt_get_publish_data(uint8_t *buffer, uint32_t *length);
uint16_t mqtt_get_id(uint8_t *buffer, uint16_t length);
int mqtt_has_valid_msg_hdr(uint8_t *buffer, uint16_t length);
mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info);
mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id);
mqtt_message_t* mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id);
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_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);
mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection);
mqtt_message_t *mqtt_msg_connect(mqtt_connection_t *connection, mqtt_connect_info_t *info);
mqtt_message_t *mqtt_msg_publish(mqtt_connection_t *connection, const char *topic, const char *data, int data_length, int qos, int retain, uint16_t *message_id);
mqtt_message_t *mqtt_msg_puback(mqtt_connection_t *connection, uint16_t message_id);
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_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);
mqtt_message_t *mqtt_msg_disconnect(mqtt_connection_t *connection);
#ifdef __cplusplus

24
lib/include/mqtt_outbox.h
View File

@ -13,18 +13,18 @@ extern "C" {
struct outbox_item;
typedef struct outbox_list_t * outbox_handle_t;
typedef struct outbox_item * outbox_item_handle_t;
typedef struct outbox_message * outbox_message_handle_t;
typedef struct outbox_list_t *outbox_handle_t;
typedef struct outbox_item *outbox_item_handle_t;
typedef struct outbox_message *outbox_message_handle_t;
typedef struct outbox_message {
uint8_t *data;
int len;
int msg_id;
int msg_qos;
int msg_type;
uint8_t *remaining_data;
int remaining_len;
uint8_t *data;
int len;
int msg_id;
int msg_qos;
int msg_type;
uint8_t *remaining_data;
int remaining_len;
} outbox_message_t;
typedef enum pending_state {
@ -37,11 +37,11 @@ outbox_handle_t outbox_init();
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handle_t message, int tick);
outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pending, int *tick);
outbox_item_handle_t outbox_get(outbox_handle_t outbox, int msg_id);
uint8_t* outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos);
uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos);
esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type);
esp_err_t outbox_delete_msgid(outbox_handle_t outbox, int msg_id);
esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type);
esp_err_t outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout);
int outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout);
esp_err_t outbox_set_pending(outbox_handle_t outbox, int msg_id, pending_state_t pending);
int outbox_get_size(outbox_handle_t outbox);

371
lib/mqtt_msg.c
View File

@ -37,8 +37,7 @@
#define MQTT_MAX_FIXED_HEADER_SIZE 5
enum mqtt_connect_flag
{
enum mqtt_connect_flag {
MQTT_CONNECT_FLAG_USERNAME = 1 << 7,
MQTT_CONNECT_FLAG_PASSWORD = 1 << 6,
MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5,
@ -46,8 +45,7 @@ enum mqtt_connect_flag
MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1
};
struct __attribute((__packed__)) mqtt_connect_variable_header
{
struct __attribute((__packed__)) mqtt_connect_variable_header {
uint8_t lengthMsb;
uint8_t lengthLsb;
#if defined(MQTT_PROTOCOL_311)
@ -61,10 +59,11 @@ struct __attribute((__packed__)) mqtt_connect_variable_header
uint8_t keepaliveLsb;
};
static int append_string(mqtt_connection_t* connection, const char* string, int len)
static int append_string(mqtt_connection_t *connection, const char *string, int len)
{
if (connection->message.length + len + 2 > connection->buffer_length)
if (connection->message.length + len + 2 > connection->buffer_length) {
return -1;
}
connection->buffer[connection->message.length++] = len >> 8;
connection->buffer[connection->message.length++] = len & 0xff;
@ -74,7 +73,7 @@ static int append_string(mqtt_connection_t* connection, const char* string, int
return len + 2;
}
static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t message_id)
static uint16_t append_message_id(mqtt_connection_t *connection, uint16_t message_id)
{
// If message_id is zero then we should assign one, otherwise
// we'll use the one supplied by the caller
@ -82,8 +81,9 @@ static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t messag
message_id = platform_random(65535);
}
if (connection->message.length + 2 > connection->buffer_length)
if (connection->message.length + 2 > connection->buffer_length) {
return 0;
}
connection->buffer[connection->message.length++] = message_id >> 8;
connection->buffer[connection->message.length++] = message_id & 0xff;
@ -91,20 +91,20 @@ static uint16_t append_message_id(mqtt_connection_t* connection, uint16_t messag
return message_id;
}
static int init_message(mqtt_connection_t* connection)
static int init_message(mqtt_connection_t *connection)
{
connection->message.length = MQTT_MAX_FIXED_HEADER_SIZE;
return MQTT_MAX_FIXED_HEADER_SIZE;
}
static mqtt_message_t* fail_message(mqtt_connection_t* connection)
static mqtt_message_t *fail_message(mqtt_connection_t *connection)
{
connection->message.data = connection->buffer;
connection->message.length = 0;
return &connection->message;
}
static mqtt_message_t* fini_message(mqtt_connection_t* connection, int type, int dup, int qos, int retain)
static mqtt_message_t *fini_message(mqtt_connection_t *connection, int type, int dup, int qos, int retain)
{
int message_length = connection->message.length - MQTT_MAX_FIXED_HEADER_SIZE;
int total_length = message_length;
@ -140,101 +140,101 @@ static mqtt_message_t* fini_message(mqtt_connection_t* connection, int type, int
// type byte
connection->buffer[offs++] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1);
// length bytes
for (int j = 0; j<len_bytes; j++) {
for (int j = 0; j < len_bytes; j++) {
connection->buffer[offs++] = encoded_lens[j];
}
return &connection->message;
}
void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length)
void mqtt_msg_init(mqtt_connection_t *connection, uint8_t *buffer, uint16_t buffer_length)
{
memset(connection, 0, sizeof(mqtt_connection_t));
connection->buffer = buffer;
connection->buffer_length = buffer_length;
}
uint32_t mqtt_get_total_length(uint8_t* buffer, uint16_t length)
uint32_t mqtt_get_total_length(uint8_t *buffer, uint16_t length, int *fixed_size_len)
{
int i;
uint32_t totlen = 0;
for (i = 1; i < length; ++i)
{
for (i = 1; i < length; ++i) {
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
totlen += i;
if (fixed_size_len) {
*fixed_size_len = i;
}
return totlen;
}
bool mqtt_header_complete(uint8_t* buffer, uint16_t buffer_length)
bool mqtt_header_complete(uint8_t *buffer, uint16_t buffer_length)
{
uint16_t i;
uint16_t topiclen;
for (i = 1; i < MQTT_MAX_FIXED_HEADER_SIZE; ++i)
{
if(i >= buffer_length)
for (i = 1; i < MQTT_MAX_FIXED_HEADER_SIZE; ++i) {
if (i >= buffer_length) {
return false;
if ((buffer[i] & 0x80) == 0)
{
}
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
// i is now the length of the fixed header
if (i + 2 >= buffer_length)
if (i + 2 >= buffer_length) {
return false;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
i += topiclen;
if (mqtt_get_qos(buffer) > 0)
{
if (mqtt_get_qos(buffer) > 0) {
i += 2;
}
// i is now the length of the fixed + variable header
return buffer_length >= i;
}
const char* mqtt_get_publish_topic(uint8_t* buffer, uint32_t* length)
char *mqtt_get_publish_topic(uint8_t *buffer, uint32_t *length)
{
int i;
int totlen = 0;
int topiclen;
for (i = 1; i < *length; ++i)
{
for (i = 1; i < *length; ++i) {
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
totlen += i;
if (i + 2 >= *length)
if (i + 2 >= *length) {
return NULL;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen > *length)
if (i + topiclen > *length) {
return NULL;
}
*length = topiclen;
return (const char*)(buffer + i);
}
return (char *)(buffer + i);
}
const char* mqtt_get_publish_data(uint8_t* buffer, uint32_t* length)
char *mqtt_get_publish_data(uint8_t *buffer, uint32_t *length)
{
int i;
int totlen = 0;
@ -242,115 +242,118 @@ const char* mqtt_get_publish_data(uint8_t* buffer, uint32_t* length)
int blength = *length;
*length = 0;
for (i = 1; i < blength; ++i)
{
for (i = 1; i < blength; ++i) {
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
if ((buffer[i] & 0x80) == 0)
{
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
totlen += i;
if (i + 2 >= blength)
if (i + 2 >= blength) {
return NULL;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen >= blength)
if (i + topiclen >= blength) {
return NULL;
}
i += topiclen;
if (mqtt_get_qos(buffer) > 0)
{
if (i + 2 >= blength)
if (mqtt_get_qos(buffer) > 0) {
if (i + 2 >= blength) {
return NULL;
}
i += 2;
}
if (totlen < i)
if (totlen < i) {
return NULL;
}
if (totlen <= blength)
if (totlen <= blength) {
*length = totlen - i;
else
} else {
*length = blength - i;
return (const char*)(buffer + i);
}
return (char *)(buffer + i);
}
uint16_t mqtt_get_id(uint8_t* buffer, uint16_t length)
uint16_t mqtt_get_id(uint8_t *buffer, uint16_t length)
{
if (length < 1)
if (length < 1) {
return 0;
}
switch (mqtt_get_type(buffer))
{
case MQTT_MSG_TYPE_PUBLISH:
{
int i;
int topiclen;
switch (mqtt_get_type(buffer)) {
case MQTT_MSG_TYPE_PUBLISH: {
int i;
int topiclen;
for (i = 1; i < length; ++i)
{
if ((buffer[i] & 0x80) == 0)
{
++i;
break;
}
}
if (i + 2 >= length)
return 0;
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen >= length)
return 0;
i += topiclen;
if (mqtt_get_qos(buffer) > 0)
{
if (i + 2 >= length)
return 0;
//i += 2;
} else {
return 0;
}
return (buffer[i] << 8) | buffer[i + 1];
}
case MQTT_MSG_TYPE_PUBACK:
case MQTT_MSG_TYPE_PUBREC:
case MQTT_MSG_TYPE_PUBREL:
case MQTT_MSG_TYPE_PUBCOMP:
case MQTT_MSG_TYPE_SUBACK:
case MQTT_MSG_TYPE_UNSUBACK:
case MQTT_MSG_TYPE_SUBSCRIBE:
{
// This requires the remaining length to be encoded in 1 byte,
// which it should be.
if (length >= 4 && (buffer[1] & 0x80) == 0)
return (buffer[2] << 8) | buffer[3];
else
return 0;
for (i = 1; i < length; ++i) {
if ((buffer[i] & 0x80) == 0) {
++i;
break;
}
}
default:
if (i + 2 >= length) {
return 0;
}
topiclen = buffer[i++] << 8;
topiclen |= buffer[i++];
if (i + topiclen > length) {
return 0;
}
i += topiclen;
if (mqtt_get_qos(buffer) > 0) {
if (i + 2 > length) {
return 0;
}
//i += 2;
} else {
return 0;
}
return (buffer[i] << 8) | buffer[i + 1];
}
case MQTT_MSG_TYPE_PUBACK:
case MQTT_MSG_TYPE_PUBREC:
case MQTT_MSG_TYPE_PUBREL:
case MQTT_MSG_TYPE_PUBCOMP:
case MQTT_MSG_TYPE_SUBACK:
case MQTT_MSG_TYPE_UNSUBACK:
case MQTT_MSG_TYPE_SUBSCRIBE:
case MQTT_MSG_TYPE_UNSUBSCRIBE: {
// This requires the remaining length to be encoded in 1 byte,
// which it should be.
if (length >= 4 && (buffer[1] & 0x80) == 0) {
return (buffer[2] << 8) | buffer[3];
} else {
return 0;
}
}
default:
return 0;
}
}
mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info)
mqtt_message_t *mqtt_msg_connect(mqtt_connection_t *connection, mqtt_connect_info_t *info)
{
struct mqtt_connect_variable_header* variable_header;
struct mqtt_connect_variable_header *variable_header;
init_message(connection);
if (connection->message.length + sizeof(*variable_header) > connection->buffer_length)
if (connection->message.length + sizeof(*variable_header) > connection->buffer_length) {
return fail_message(connection);
variable_header = (void*)(connection->buffer + connection->message.length);
}
variable_header = (void *)(connection->buffer + connection->message.length);
connection->message.length += sizeof(*variable_header);
variable_header->lengthMsb = 0;
@ -368,43 +371,46 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
variable_header->keepaliveMsb = info->keepalive >> 8;
variable_header->keepaliveLsb = info->keepalive & 0xff;
if (info->clean_session)
if (info->clean_session) {
variable_header->flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION;
if (info->client_id != NULL && info->client_id[0] != '\0')
{
if (append_string(connection, info->client_id, strlen(info->client_id)) < 0)
return fail_message(connection);
}
else
if (info->client_id != NULL && info->client_id[0] != '\0') {
if (append_string(connection, info->client_id, strlen(info->client_id)) < 0) {
return fail_message(connection);
}
} else {
return fail_message(connection);
}
if (info->will_topic != NULL && info->will_topic[0] != '\0')
{
if (append_string(connection, info->will_topic, strlen(info->will_topic)) < 0)
if (info->will_topic != NULL && info->will_topic[0] != '\0') {
if (append_string(connection, info->will_topic, strlen(info->will_topic)) < 0) {
return fail_message(connection);
}
if (append_string(connection, info->will_message, info->will_length) < 0)
if (append_string(connection, info->will_message, info->will_length) < 0) {
return fail_message(connection);
}
variable_header->flags |= MQTT_CONNECT_FLAG_WILL;
if (info->will_retain)
if (info->will_retain) {
variable_header->flags |= MQTT_CONNECT_FLAG_WILL_RETAIN;
}
variable_header->flags |= (info->will_qos & 3) << 3;
}
if (info->username != NULL && info->username[0] != '\0')
{
if (append_string(connection, info->username, strlen(info->username)) < 0)
if (info->username != NULL && info->username[0] != '\0') {
if (append_string(connection, info->username, strlen(info->username)) < 0) {
return fail_message(connection);
}
variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME;
}
if (info->password != NULL && info->password[0] != '\0')
{
if (append_string(connection, info->password, strlen(info->password)) < 0)
if (info->password != NULL && info->password[0] != '\0') {
if (append_string(connection, info->password, strlen(info->password)) < 0) {
return fail_message(connection);
}
variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD;
}
@ -412,23 +418,25 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
return fini_message(connection, MQTT_MSG_TYPE_CONNECT, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id)
mqtt_message_t *mqtt_msg_publish(mqtt_connection_t *connection, const char *topic, const char *data, int data_length, int qos, int retain, uint16_t *message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0')
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
if (append_string(connection, topic, strlen(topic)) < 0)
return fail_message(connection);
if (qos > 0)
{
if ((*message_id = append_message_id(connection, 0)) == 0)
return fail_message(connection);
}
else
if (append_string(connection, topic, strlen(topic)) < 0) {
return fail_message(connection);
}
if (qos > 0) {
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
} else {
*message_id = 0;
}
if (connection->message.length + data_length > connection->buffer_length) {
// Not enough size in buffer -> fragment this message
@ -444,88 +452,139 @@ mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topi
return fini_message(connection, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain);
}
mqtt_message_t* mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id)
mqtt_message_t *mqtt_msg_puback(mqtt_connection_t *connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
if (append_message_id(connection, message_id) == 0) {
return fail_message(connection);
}
return fini_message(connection, MQTT_MSG_TYPE_PUBACK, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id)
mqtt_message_t *mqtt_msg_pubrec(mqtt_connection_t *connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
if (append_message_id(connection, message_id) == 0) {
return fail_message(connection);
}
return fini_message(connection, MQTT_MSG_TYPE_PUBREC, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id)
mqtt_message_t *mqtt_msg_pubrel(mqtt_connection_t *connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
if (append_message_id(connection, message_id) == 0) {
return fail_message(connection);
}
return fini_message(connection, MQTT_MSG_TYPE_PUBREL, 0, 1, 0);
}
mqtt_message_t* mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id)
mqtt_message_t *mqtt_msg_pubcomp(mqtt_connection_t *connection, uint16_t message_id)
{
init_message(connection);
if (append_message_id(connection, message_id) == 0)
if (append_message_id(connection, message_id) == 0) {
return fail_message(connection);
}
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 char *topic, int qos, uint16_t *message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0')
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
}
if ((*message_id = append_message_id(connection, 0)) == 0)
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
if (append_string(connection, topic, strlen(topic)) < 0)
if (append_string(connection, topic, strlen(topic)) < 0) {
return fail_message(connection);
}
if (connection->message.length + 1 > connection->buffer_length)
if (connection->message.length + 1 > connection->buffer_length) {
return fail_message(connection);
}
connection->buffer[connection->message.length++] = qos;
return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0);
}
mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id)
mqtt_message_t *mqtt_msg_unsubscribe(mqtt_connection_t *connection, const char *topic, uint16_t *message_id)
{
init_message(connection);
if (topic == NULL || topic[0] == '\0')
if (topic == NULL || topic[0] == '\0') {
return fail_message(connection);
}
if ((*message_id = append_message_id(connection, 0)) == 0)
if ((*message_id = append_message_id(connection, 0)) == 0) {
return fail_message(connection);
}
if (append_string(connection, topic, strlen(topic)) < 0)
if (append_string(connection, topic, strlen(topic)) < 0) {
return fail_message(connection);
}
return fini_message(connection, MQTT_MSG_TYPE_UNSUBSCRIBE, 0, 1, 0);
}
mqtt_message_t* mqtt_msg_pingreq(mqtt_connection_t* connection)
mqtt_message_t *mqtt_msg_pingreq(mqtt_connection_t *connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_pingresp(mqtt_connection_t* connection)
mqtt_message_t *mqtt_msg_pingresp(mqtt_connection_t *connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_PINGRESP, 0, 0, 0);
}
mqtt_message_t* mqtt_msg_disconnect(mqtt_connection_t* connection)
mqtt_message_t *mqtt_msg_disconnect(mqtt_connection_t *connection)
{
init_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_DISCONNECT, 0, 0, 0);
}
/*
* check flags: [MQTT-2.2.2-1], [MQTT-2.2.2-2]
* returns 0 if flags are invalid, otherwise returns 1
*/
int mqtt_has_valid_msg_hdr(uint8_t *buffer, uint16_t length)
{
int qos, dup;
if (length < 1) {
return 0;
}
switch (mqtt_get_type(buffer)) {
case MQTT_MSG_TYPE_CONNECT:
case MQTT_MSG_TYPE_CONNACK:
case MQTT_MSG_TYPE_PUBACK:
case MQTT_MSG_TYPE_PUBREC:
case MQTT_MSG_TYPE_PUBCOMP:
case MQTT_MSG_TYPE_SUBACK:
case MQTT_MSG_TYPE_UNSUBACK:
case MQTT_MSG_TYPE_PINGREQ:
case MQTT_MSG_TYPE_PINGRESP:
case MQTT_MSG_TYPE_DISCONNECT:
return (buffer[0] & 0x0f) == 0; /* all flag bits are 0 */
case MQTT_MSG_TYPE_PUBREL:
case MQTT_MSG_TYPE_SUBSCRIBE:
case MQTT_MSG_TYPE_UNSUBSCRIBE:
return (buffer[0] & 0x0f) == 0x02; /* only bit 1 is set */
case MQTT_MSG_TYPE_PUBLISH:
qos = mqtt_get_qos(buffer);
dup = mqtt_get_dup(buffer);
/*
* there is no qos=3 [MQTT-3.3.1-4]
* dup flag must be set to 0 for all qos=0 messages [MQTT-3.3.1-2]
*/
return (qos < 3) && ((qos > 0) || (dup == 0));
default:
return 0;
}
}

17
lib/mqtt_outbox.c
View File

@ -49,7 +49,7 @@ outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, outbox_message_handl
});
memcpy(item->buffer, message->data, message->len);
if (message->remaining_data) {
memcpy(item->buffer+message->len, message->remaining_data, message->remaining_len);
memcpy(item->buffer + message->len, message->remaining_data, message->remaining_len);
}
STAILQ_INSERT_TAIL(outbox, item, next);
ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%d", message->msg_id, message->msg_type, message->len + message->remaining_len, outbox_get_size(outbox));
@ -81,14 +81,14 @@ outbox_item_handle_t outbox_dequeue(outbox_handle_t outbox, pending_state_t pend
return NULL;
}
uint8_t* outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos)
uint8_t *outbox_item_get_data(outbox_item_handle_t item, size_t *len, uint16_t *msg_id, int *msg_type, int *qos)
{
if (item) {
*len = item->len;
*msg_id = item->msg_id;
*msg_type = item->msg_type;
*qos = item->msg_qos;
return (uint8_t*)item->buffer;
return (uint8_t *)item->buffer;
}
return NULL;
}
@ -97,7 +97,7 @@ esp_err_t outbox_delete(outbox_handle_t outbox, int msg_id, int msg_type)
{
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (item->msg_id == msg_id && (0xFF&(item->msg_type)) == msg_type) {
if (item->msg_id == msg_id && (0xFF & (item->msg_type)) == msg_type) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
free(item->buffer);
free(item);
@ -145,18 +145,21 @@ esp_err_t outbox_delete_msgtype(outbox_handle_t outbox, int msg_type)
return ESP_OK;
}
esp_err_t outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout)
int outbox_delete_expired(outbox_handle_t outbox, int current_tick, int timeout)
{
int deleted_items = 0;
outbox_item_handle_t item, tmp;
STAILQ_FOREACH_SAFE(item, outbox, next, tmp) {
if (current_tick - item->tick > timeout) {
STAILQ_REMOVE(outbox, item, outbox_item, next);
printf("free message\n");
free(item->buffer);
free(item);
deleted_items ++;
}
}
return ESP_OK;
return deleted_items;
}
int outbox_get_size(outbox_handle_t outbox)
@ -171,7 +174,7 @@ int outbox_get_size(outbox_handle_t outbox)
esp_err_t outbox_cleanup(outbox_handle_t outbox, int max_size)
{
while(outbox_get_size(outbox) > max_size) {
while (outbox_get_size(outbox) > max_size) {
outbox_item_handle_t item = outbox_dequeue(outbox, CONFIRMED, NULL);
if (item == NULL) {
return ESP_FAIL;

4
lib/platform_esp32_idf.c
View File

@ -21,14 +21,14 @@ char *platform_create_id_string()
int platform_random(int max)
{
return esp_random()%max;
return esp_random() % max;
}
long long platform_tick_get_ms()
{
struct timeval te;
gettimeofday(&te, NULL); // get current time
long long milliseconds = te.tv_sec*1000LL + te.tv_usec/1000; // calculate milliseconds
long long milliseconds = te.tv_sec * 1000LL + te.tv_usec / 1000; // calculate milliseconds
// printf("milliseconds: %lld\n", milliseconds);
return milliseconds;
}

8
modify_for_legacy_idf.sh
View File

@ -1,5 +1,11 @@
#!/usr/bin/env bash
if [[ -z $1 ]]; then
LATEST_IDF=master
else
LATEST_IDF=$1
fi
# This snipped prepares environment for using esp-mqtt repository separately from idf -- legacy use before IDFv3.2
#
esp_mqtt_path=`pwd`
@ -7,7 +13,7 @@ mkdir -p ${esp_mqtt_path}/examples
pushd
cd $IDF_PATH
former_commit_id=`git rev-parse HEAD`
git checkout master
git checkout ${LATEST_IDF}
for example in tcp; do
cp -r $IDF_PATH/examples/protocols/mqtt/${example} ${esp_mqtt_path}/examples

753
mqtt_client.c
View File

@ -27,15 +27,14 @@
static const char *TAG = "MQTT_CLIENT";
typedef struct mqtt_state
{
typedef struct mqtt_state {
mqtt_connect_info_t *connect_info;
uint8_t *in_buffer;
uint8_t *out_buffer;
int in_buffer_length;
int out_buffer_length;
uint32_t message_length;
uint32_t message_length_read;
uint32_t in_buffer_read_len;
mqtt_message_t *outbound_message;
mqtt_connection_t mqtt_connection;
uint16_t pending_msg_id;
@ -104,6 +103,7 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
static esp_err_t esp_mqtt_abort_connection(esp_mqtt_client_handle_t client);
static esp_err_t esp_mqtt_client_ping(esp_mqtt_client_handle_t client);
static char *create_string(const char *ptr, int len);
static int mqtt_message_receive(esp_mqtt_client_handle_t client, int read_poll_timeout_ms);
esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config)
{
@ -267,19 +267,21 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
client->mqtt_state.pending_msg_id);
write_len = esp_transport_write(client->transport,
(char *)client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->config->network_timeout_ms);
(char *)client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->config->network_timeout_ms);
if (write_len < 0) {
ESP_LOGE(TAG, "Writing failed, errno= %d", errno);
return ESP_FAIL;
}
read_len = esp_transport_read(client->transport,
(char *)client->mqtt_state.in_buffer,
client->mqtt_state.in_buffer_length,
client->config->network_timeout_ms);
if (read_len < 0) {
ESP_LOGE(TAG, "Error network response");
client->mqtt_state.in_buffer_read_len = 0;
client->mqtt_state.message_length = 0;
/* wait configured network timeout for broker connection response */
read_len = mqtt_message_receive(client, client->config->network_timeout_ms);
if (read_len <= 0) {
ESP_LOGE(TAG, "%s: mqtt_message_receive() returned %d", __func__, read_len);
return ESP_FAIL;
}
@ -289,24 +291,24 @@ static esp_err_t esp_mqtt_connect(esp_mqtt_client_handle_t client, int timeout_m
}
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
switch (connect_rsp_code) {
case CONNECTION_ACCEPTED:
ESP_LOGD(TAG, "Connected");
return ESP_OK;
case CONNECTION_REFUSE_PROTOCOL:
ESP_LOGW(TAG, "Connection refused, bad protocol");
return ESP_FAIL;
case CONNECTION_REFUSE_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Connection refused, server unavailable");
return ESP_FAIL;
case CONNECTION_REFUSE_BAD_USERNAME:
ESP_LOGW(TAG, "Connection refused, bad username or password");
return ESP_FAIL;
case CONNECTION_REFUSE_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Connection refused, not authorized");
return ESP_FAIL;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
return ESP_FAIL;
case CONNECTION_ACCEPTED:
ESP_LOGD(TAG, "Connected");
return ESP_OK;
case CONNECTION_REFUSE_PROTOCOL:
ESP_LOGW(TAG, "Connection refused, bad protocol");
return ESP_FAIL;
case CONNECTION_REFUSE_SERVER_UNAVAILABLE:
ESP_LOGW(TAG, "Connection refused, server unavailable");
return ESP_FAIL;
case CONNECTION_REFUSE_BAD_USERNAME:
ESP_LOGW(TAG, "Connection refused, bad username or password");
return ESP_FAIL;
case CONNECTION_REFUSE_NOT_AUTHORIZED:
ESP_LOGW(TAG, "Connection refused, not authorized");
return ESP_FAIL;
default:
ESP_LOGW(TAG, "Connection refused, Unknow reason");
return ESP_FAIL;
}
return ESP_OK;
}
@ -419,17 +421,25 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
return client;
_mqtt_init_failed:
esp_mqtt_client_destroy(client);
MQTT_API_UNLOCK(client);
return NULL;
}
esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client)
{
if (client == NULL) {
return ESP_ERR_INVALID_ARG;
}
esp_mqtt_client_stop(client);
esp_mqtt_destroy_config(client);
esp_transport_list_destroy(client->transport_list);
outbox_destroy(client->outbox);
vEventGroupDelete(client->status_bits);
if (client->transport_list) {
esp_transport_list_destroy(client->transport_list);
}
if (client->outbox) {
outbox_destroy(client->outbox);
}
if (client->status_bits) {
vEventGroupDelete(client->status_bits);
}
free(client->mqtt_state.in_buffer);
free(client->mqtt_state.out_buffer);
vSemaphoreDelete(client->api_lock);
@ -480,7 +490,7 @@ esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *u
}
if (puri.field_data[UF_PORT].len) {
client->config->port = strtol((const char*)(uri + puri.field_data[UF_PORT].off), NULL, 10);
client->config->port = strtol((const char *)(uri + puri.field_data[UF_PORT].off), NULL, 10);
}
char *user_info = create_string(uri + puri.field_data[UF_USERINFO].off, puri.field_data[UF_USERINFO].len);
@ -502,9 +512,9 @@ esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *u
static esp_err_t mqtt_write_data(esp_mqtt_client_handle_t client)
{
int write_len = esp_transport_write(client->transport,
(char *)client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->config->network_timeout_ms);
(char *)client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length,
client->config->network_timeout_ms);
// client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
if (write_len <= 0) {
ESP_LOGE(TAG, "Error write data or timeout, written len = %d, errno=%d", write_len, errno);
@ -534,92 +544,65 @@ static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client)
return ESP_FAIL;
}
static void deliver_publish(esp_mqtt_client_handle_t client, uint8_t *message, int length)
static esp_err_t deliver_publish(esp_mqtt_client_handle_t client)
{
mqtt_header_state_t mqtt_header_state = MQTT_HEADER_STATE_INCOMPLETE;
const char *mqtt_topic = NULL, *mqtt_data = NULL;
uint32_t mqtt_topic_length, mqtt_data_length;
uint32_t mqtt_len = 0, mqtt_offset = 0;
int len_read= length;
int max_to_read = client->mqtt_state.in_buffer_length;
int buffer_offset = 0;
esp_transport_handle_t transport = client->transport;
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;
size_t msg_topic_len = msg_read_len, msg_data_len = msg_read_len;
size_t msg_data_offset = 0;
char *msg_topic = NULL, *msg_data = NULL;
do
{
if (mqtt_header_state == MQTT_HEADER_STATE_INCOMPLETE) {
// any further reading only the underlying payload
transport = esp_transport_get_payload_transport_handle(transport);
if (!mqtt_header_complete(message, length)) {
// mqtt header is not complete, continue reading
memmove(client->mqtt_state.in_buffer, message, length);
buffer_offset = length;
message = client->mqtt_state.in_buffer;
max_to_read = client->mqtt_state.in_buffer_length - length;
mqtt_len = 0;
} else {
// mqtt header was fully read
mqtt_header_state = MQTT_HEADER_STATE_COMPLETE;
mqtt_data_length = mqtt_topic_length = length;
// Read the topic
if(NULL == (mqtt_topic = mqtt_get_publish_topic(message, &mqtt_topic_length))) {
ESP_LOGE(TAG, "Unable to read topic from header");
break;
}
// Read the payload
mqtt_data = mqtt_get_publish_data(message, &mqtt_data_length);
if(mqtt_data_length > 0 && mqtt_data == NULL) {
ESP_LOGE(TAG, "Unable to read data from message");
break;
} else if(mqtt_data_length > 0) {
mqtt_len = mqtt_data_length;
mqtt_data_length = client->mqtt_state.message_length - ((uint8_t*)mqtt_data- message);
} else {
mqtt_len = 0;
}
// read msg id only once
client->event.msg_id = mqtt_get_id(message, length);
}
} else {
mqtt_len = len_read;
mqtt_data = (const char*)client->mqtt_state.in_buffer;
mqtt_topic = NULL;
mqtt_topic_length = 0;
// get topic
msg_topic = mqtt_get_publish_topic(msg_buf, &msg_topic_len);
if (msg_topic == NULL) {
ESP_LOGE(TAG, "%s: mqtt_get_publish_topic() failed", __func__);
return ESP_FAIL;
}
ESP_LOGD(TAG, "%s: msg_topic_len=%u", __func__, msg_topic_len);
// get payload
msg_data = mqtt_get_publish_data(msg_buf, &msg_data_len);
if (msg_data_len > 0 && msg_data == NULL) {
ESP_LOGE(TAG, "%s: mqtt_get_publish_data() failed", __func__);
return ESP_FAIL;
}
// post data event
client->event.msg_id = mqtt_get_id(msg_buf, msg_data_len);
client->event.total_data_len = msg_data_len + msg_total_len - msg_read_len;
post_data_event:
ESP_LOGD(TAG, "Get data len= %d, topic len=%d, total_data: %d offset: %d", msg_data_len, msg_topic_len,
client->event.total_data_len, msg_data_offset);
client->event.event_id = MQTT_EVENT_DATA;
client->event.data = msg_data_len > 0 ? msg_data : NULL;
client->event.data_len = msg_data_len;
client->event.current_data_offset = msg_data_offset;
client->event.topic = msg_topic;
client->event.topic_len = msg_topic_len;
esp_mqtt_dispatch_event(client);
if (msg_read_len < msg_total_len) {
// if total data is longer then actual -> read payload only
size_t buf_len = client->mqtt_state.in_buffer_length;
esp_transport_handle_t transport = esp_transport_get_payload_transport_handle(client->transport);
msg_data = (char *)client->mqtt_state.in_buffer;
msg_topic = NULL;
msg_topic_len = 0;
msg_data_offset += msg_data_len;
msg_data_len = esp_transport_read(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);
if (msg_data_len <= 0) {
ESP_LOGE(TAG, "Read error or timeout: len_read=%d, errno=%d", msg_data_len, errno);
return ESP_FAIL;
}
if (mqtt_header_state == MQTT_HEADER_STATE_COMPLETE) {
ESP_LOGD(TAG, "Get data len= %d, topic len=%d", mqtt_len, mqtt_topic_length);
client->event.event_id = MQTT_EVENT_DATA;
client->event.data = mqtt_len > 0 ? (char *)mqtt_data : NULL;
client->event.data_len = mqtt_len;
client->event.total_data_len = mqtt_data_length;
client->event.current_data_offset = mqtt_offset;
client->event.topic = (char *)mqtt_topic;
client->event.topic_len = mqtt_topic_length;
esp_mqtt_dispatch_event(client);
}
mqtt_offset += mqtt_len;
if (client->mqtt_state.message_length_read >= client->mqtt_state.message_length) {
break;
}
len_read = esp_transport_read(transport,
(char *)client->mqtt_state.in_buffer + buffer_offset,
client->mqtt_state.message_length - client->mqtt_state.message_length_read > max_to_read ?
max_to_read : client->mqtt_state.message_length - client->mqtt_state.message_length_read,
client->config->network_timeout_ms);
length = len_read + buffer_offset;
buffer_offset = 0;
max_to_read = client->mqtt_state.in_buffer_length;
if (len_read <= 0) {
ESP_LOGE(TAG, "Read error or timeout: len_read=%d, errno=%d", len_read, errno);
break;
}
client->mqtt_state.message_length_read += len_read;
} while (1);
msg_read_len += msg_data_len;
goto post_data_event;
}
return ESP_OK;
}
static bool is_valid_mqtt_msg(esp_mqtt_client_handle_t client, int msg_type, int msg_id)
@ -677,109 +660,237 @@ static void mqtt_enqueue(esp_mqtt_client_handle_t client)
//unlock
}
/*
* Returns:
* -1 in case of failure
* 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
*/
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;
esp_transport_handle_t t = client->transport;
client->mqtt_state.message_length = 0;
if (client->mqtt_state.in_buffer_read_len == 0) {
/*
* Read first byte of the mqtt packet fixed header, it contains packet
* type and flags.
*/
read_len = esp_transport_read(t, (char *)buf, 1, read_poll_timeout_ms);
if (read_len < 0) {
ESP_LOGE(TAG, "%s: transport_read() error: errno=%d", __func__, errno);
goto err;
}
if (read_len == 0) {
ESP_LOGV(TAG, "%s: transport_read(): no data or EOF", __func__);
return 0;
}
ESP_LOGD(TAG, "%s: first byte: 0x%x", __func__, *buf);
/*
* Verify the flags and act according to MQTT protocol: close connection
* if the flags are set incorrectly.
*/
if (!mqtt_has_valid_msg_hdr(buf, read_len)) {
ESP_LOGE(TAG, "%s: received a message with an invalid header=0x%x", __func__, *buf);
goto err;
}
buf++;
client->mqtt_state.in_buffer_read_len++;
}
/* any further reading only the underlying payload */
t = esp_transport_get_payload_transport_handle(t);
if ((client->mqtt_state.in_buffer_read_len == 1) ||
((client->mqtt_state.in_buffer_read_len < 6) && (*(buf - 1) & 0x80))) {
do {
/*
* Read the "remaining length" part of mqtt packet fixed header. It
* starts at second byte and spans up to 4 bytes, but we accept here
* only up to 2 bytes of remaining length, i.e. messages with
* maximal remaining length value = 16383 (maximal total message
* size of 16386 bytes).
*/
read_len = esp_transport_read(t, (char *)buf, 1, read_poll_timeout_ms);
if (read_len < 0) {
ESP_LOGE(TAG, "%s: transport_read() error: errno=%d", __func__, errno);
goto err;
}
if (read_len == 0) {
ESP_LOGD(TAG, "%s: transport_read(): no data or EOF", __func__);
return 0;
}
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));
}
total_len = mqtt_get_total_length(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_read_len, &fixed_header_len);
ESP_LOGD(TAG, "%s: total message length: %d (already read: %u)", __func__, total_len, client->mqtt_state.in_buffer_read_len);
client->mqtt_state.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) {
/*
* 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) {
/* 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_poll_timeout_ms);
ESP_LOGD(TAG, "%s: read_len=%d", __func__, read_len);
if (read_len < 0) {
ESP_LOGE(TAG, "%s: transport_read() error: errno=%d", __func__, errno);
goto err;
} else if (read_len == 0) {
ESP_LOGD(TAG, "%s: transport_read(): no data or EOF", __func__);
return 0;
}
client->mqtt_state.in_buffer_read_len += read_len;
buf += read_len;
if (client->mqtt_state.in_buffer_read_len < fixed_header_len + 2) {
ESP_LOGD(TAG, "%s: transport_read(): message reading left in progress :: total message length: %d (already read: %u)",
__func__, total_len, client->mqtt_state.in_buffer_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);
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) {
ESP_LOGE(TAG, "%s: message is too big, insufficient buffer size", __func__);
goto err;
} else {
total_len = client->mqtt_state.in_buffer_length;
}
/* free to continue with reading */
} else {
ESP_LOGE(TAG, "%s: message is too big, insufficient buffer size", __func__);
goto err;
}
}
if (client->mqtt_state.in_buffer_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);
ESP_LOGD(TAG, "%s: read_len=%d", __func__, read_len);
if (read_len < 0) {
ESP_LOGE(TAG, "%s: transport_read() error: errno=%d", __func__, errno);
goto err;
}
if (read_len == 0) {
ESP_LOGD(TAG, "%s: transport_read(): no data or EOF", __func__);
return 0;
}
client->mqtt_state.in_buffer_read_len += read_len;
if (client->mqtt_state.in_buffer_read_len < total_len) {
ESP_LOGD(TAG, "%s: transport_read(): message reading left in progress :: total message length: %d (already read: %u)",
__func__, total_len, client->mqtt_state.in_buffer_read_len);
return 0;
}
}
ESP_LOGD(TAG, "%s: transport_read():%d %d", __func__, client->mqtt_state.in_buffer_read_len, client->mqtt_state.message_length);
return 1;
err:
return -1;
}
static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
{
int read_len;
uint8_t msg_type;
uint8_t msg_qos;
uint8_t msg_type;
uint8_t msg_qos;
uint16_t msg_id;
uint32_t transport_message_offset = 0 ;
read_len = esp_transport_read(client->transport, (char *)client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length, 0);
if (read_len < 0) {
ESP_LOGE(TAG, "Read error or end of stream, errno:%d", errno);
/* non-blocking receive in order not to block other tasks */
int recv = mqtt_message_receive(client, 0);
if (recv < 0) {
ESP_LOGE(TAG, "%s: mqtt_message_receive() returned %d", __func__, recv);
return ESP_FAIL;
}
if (read_len == 0) {
if (recv == 0) {
return ESP_OK;
}
int read_len = client->mqtt_state.message_length;
// In case of fragmented packet (when receiving a large publish message), the deliver_publish function will read the rest of the message with more transport read, which means the MQTT message length will be greater than the initial transport read length. That explains that the stopping condition is not the equality here
while ( transport_message_offset < read_len ) {
// 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[transport_message_offset]);
msg_qos = mqtt_get_qos(&client->mqtt_state.in_buffer[transport_message_offset]);
msg_id = mqtt_get_id(&client->mqtt_state.in_buffer[transport_message_offset], read_len - transport_message_offset);
client->mqtt_state.message_length_read = read_len - transport_message_offset;
client->mqtt_state.message_length = mqtt_get_total_length(&client->mqtt_state.in_buffer[transport_message_offset], client->mqtt_state.message_length_read);
// 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_id = mqtt_get_id(client->mqtt_state.in_buffer, read_len);
ESP_LOGD(TAG, "msg_type=%d, msg_id=%d", msg_type, msg_id);
ESP_LOGD(TAG, "msg_type=%d, msg_id=%d", msg_type, msg_id);
switch (msg_type)
{
case MQTT_MSG_TYPE_SUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_SUBSCRIBE, msg_id)) {
ESP_LOGD(TAG, "Subscribe successful");
client->event.event_id = MQTT_EVENT_SUBSCRIBED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_UNSUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_UNSUBSCRIBE, msg_id)) {
ESP_LOGD(TAG, "UnSubscribe successful");
client->event.event_id = MQTT_EVENT_UNSUBSCRIBED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PUBLISH:
if (msg_qos == 1) {
client->mqtt_state.outbound_message = mqtt_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
}
else if (msg_qos == 2) {
client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
}
if (msg_qos == 1 || msg_qos == 2) {
ESP_LOGD(TAG, "Queue response QoS: %d", msg_qos);
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error write qos msg repsonse, qos = %d", msg_qos);
// TODO: Shoule reconnect?
// return ESP_FAIL;
}
}
// Deliver the publish message
ESP_LOGD(TAG, "deliver_publish, message_length_read=%d, message_length=%d", client->mqtt_state.message_length_read, client->mqtt_state.message_length);
deliver_publish(client, &client->mqtt_state.in_buffer[transport_message_offset], client->mqtt_state.message_length_read);
break;
case MQTT_MSG_TYPE_PUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish");
outbox_set_pending(client->outbox, msg_id, CONFIRMED);
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PUBREC:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREC");
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
outbox_set_pending(client->outbox, msg_id, CONFIRMED);
mqtt_write_data(client);
break;
case MQTT_MSG_TYPE_PUBREL:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREL");
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
mqtt_write_data(client);
break;
case MQTT_MSG_TYPE_PUBCOMP:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBCOMP");
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PINGRESP:
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PINGRESP");
client->wait_for_ping_resp = false;
break;
switch (msg_type) {
case MQTT_MSG_TYPE_SUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_SUBSCRIBE, msg_id)) {
ESP_LOGD(TAG, "Subscribe successful");
client->event.event_id = MQTT_EVENT_SUBSCRIBED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_UNSUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_UNSUBSCRIBE, msg_id)) {
ESP_LOGD(TAG, "UnSubscribe successful");
client->event.event_id = MQTT_EVENT_UNSUBSCRIBED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PUBLISH:
ESP_LOGD(TAG, "deliver_publish, message_length_read=%d, message_length=%d", client->mqtt_state.in_buffer_read_len, client->mqtt_state.message_length);
if (deliver_publish(client) != ESP_OK) {
ESP_LOGE(TAG, "Failed to deliver publish message id=%d", msg_id);
return ESP_FAIL;
}
if (msg_qos == 1) {
client->mqtt_state.outbound_message = mqtt_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
} else if (msg_qos == 2) {
client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
}
transport_message_offset += client->mqtt_state.message_length;
if (msg_qos == 1 || msg_qos == 2) {
ESP_LOGD(TAG, "Queue response QoS: %d", msg_qos);
if (mqtt_write_data(client) != ESP_OK) {
ESP_LOGE(TAG, "Error write qos msg repsonse, qos = %d", msg_qos);
return ESP_FAIL;
}
}
break;
case MQTT_MSG_TYPE_PUBACK:
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish");
outbox_set_pending(client->outbox, msg_id, CONFIRMED);
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PUBREC:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREC");
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
outbox_set_pending(client->outbox, msg_id, CONFIRMED);
mqtt_write_data(client);
break;
case MQTT_MSG_TYPE_PUBREL:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBREL");
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
mqtt_write_data(client);
break;
case MQTT_MSG_TYPE_PUBCOMP:
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBCOMP");
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBLISH, msg_id)) {
ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event_with_msgid(client);
}
break;
case MQTT_MSG_TYPE_PINGRESP:
ESP_LOGD(TAG, "MQTT_MSG_TYPE_PINGRESP");
client->wait_for_ping_resp = false;
break;
}
client->mqtt_state.in_buffer_read_len = 0;
return ESP_OK;
}
@ -787,9 +898,9 @@ static esp_err_t mqtt_resend_queued(esp_mqtt_client_handle_t client, outbox_item
{
// decode queued data
client->mqtt_state.outbound_message->data = outbox_item_get_data(item, &client->mqtt_state.outbound_message->length, &client->mqtt_state.pending_msg_id,
&client->mqtt_state.pending_msg_type, &client->mqtt_state.pending_publish_qos);
&client->mqtt_state.pending_msg_type, &client->mqtt_state.pending_publish_qos);
// set duplicate flag for QoS-2 message
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH &&client->mqtt_state.pending_publish_qos==2) {
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_publish_qos == 2) {
mqtt_set_dup(client->mqtt_state.outbound_message->data);
}
@ -826,107 +937,111 @@ static void esp_mqtt_task(void *pv)
while (client->run) {
MQTT_API_LOCK(client);
switch ((int)client->state) {
case MQTT_STATE_INIT:
xEventGroupClearBits(client->status_bits, RECONNECT_BIT);
client->event.event_id = MQTT_EVENT_BEFORE_CONNECT;
esp_mqtt_dispatch_event_with_msgid(client);
case MQTT_STATE_INIT:
xEventGroupClearBits(client->status_bits, RECONNECT_BIT);
client->event.event_id = MQTT_EVENT_BEFORE_CONNECT;
esp_mqtt_dispatch_event_with_msgid(client);
if (client->transport == NULL) {
ESP_LOGE(TAG, "There are no transport");
client->run = false;
}
if (client->transport == NULL) {
ESP_LOGE(TAG, "There are no transport");
client->run = false;
}
if (esp_transport_connect(client->transport,
if (esp_transport_connect(client->transport,
client->config->host,
client->config->port,
client->config->network_timeout_ms) < 0) {
ESP_LOGE(TAG, "Error transport connect");
esp_mqtt_abort_connection(client);
break;
}
ESP_LOGD(TAG, "Transport connected to %s://%s:%d", client->config->scheme, client->config->host, client->config->port);
if (esp_mqtt_connect(client, client->config->network_timeout_ms) != ESP_OK) {
ESP_LOGI(TAG, "Error MQTT Connected");
esp_mqtt_abort_connection(client);
break;
}
client->event.event_id = MQTT_EVENT_CONNECTED;
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.in_buffer);
client->state = MQTT_STATE_CONNECTED;
esp_mqtt_dispatch_event_with_msgid(client);
client->refresh_connection_tick = platform_tick_get_ms();
ESP_LOGE(TAG, "Error transport connect");
esp_mqtt_abort_connection(client);
break;
case MQTT_STATE_CONNECTED:
// receive and process data
if (mqtt_process_receive(client) == ESP_FAIL) {
esp_mqtt_abort_connection(client);
break;
}
}
ESP_LOGD(TAG, "Transport connected to %s://%s:%d", client->config->scheme, client->config->host, client->config->port);
if (esp_mqtt_connect(client, client->config->network_timeout_ms) != ESP_OK) {
ESP_LOGI(TAG, "Error MQTT Connected");
esp_mqtt_abort_connection(client);
break;
}
client->event.event_id = MQTT_EVENT_CONNECTED;
client->event.session_present = mqtt_get_connect_session_present(client->mqtt_state.in_buffer);
client->state = MQTT_STATE_CONNECTED;
esp_mqtt_dispatch_event_with_msgid(client);
client->refresh_connection_tick = platform_tick_get_ms();
// resend all non-transmitted messages first
outbox_item_handle_t item = outbox_dequeue(client->outbox, QUEUED, NULL);
if (item) {
if (mqtt_resend_queued(client, item) == ESP_OK) {
outbox_set_pending(client->outbox, client->mqtt_state.pending_msg_id, TRANSMITTED);
}
break;
case MQTT_STATE_CONNECTED:
// receive and process data
if (mqtt_process_receive(client) == ESP_FAIL) {
esp_mqtt_abort_connection(client);
break;
}
// resend all non-transmitted messages first
outbox_item_handle_t item = outbox_dequeue(client->outbox, QUEUED, NULL);
if (item) {
if (mqtt_resend_queued(client, item) == ESP_OK) {
outbox_set_pending(client->outbox, client->mqtt_state.pending_msg_id, TRANSMITTED);
}
// resend other "transmitted" messages after 1s
} else if (platform_tick_get_ms() - last_retransmit > 1000) {
last_retransmit = platform_tick_get_ms();
item = outbox_dequeue(client->outbox, TRANSMITTED, &msg_tick);
if (item && (last_retransmit - msg_tick > 1000)) {
mqtt_resend_queued(client, item);
}
} else if (platform_tick_get_ms() - last_retransmit > 1000) {
last_retransmit = platform_tick_get_ms();
item = outbox_dequeue(client->outbox, TRANSMITTED, &msg_tick);
if (item && (last_retransmit - msg_tick > 1000)) {
mqtt_resend_queued(client, item);
}
}
if (platform_tick_get_ms() - client->keepalive_tick > client->connect_info.keepalive * 1000 / 2) {
//No ping resp from last ping => Disconnected
if(client->wait_for_ping_resp){
ESP_LOGE(TAG, "No PING_RESP, disconnected");
esp_mqtt_abort_connection(client);
client->wait_for_ping_resp = false;
break;
}
if (esp_mqtt_client_ping(client) == ESP_FAIL) {
ESP_LOGE(TAG, "Can't send ping, disconnected");
esp_mqtt_abort_connection(client);
break;
} else {
client->wait_for_ping_resp = true;
}
ESP_LOGD(TAG, "PING sent");
}
if (client->config->refresh_connection_after_ms &&
platform_tick_get_ms() - client->refresh_connection_tick > client->config->refresh_connection_after_ms) {
ESP_LOGD(TAG, "Refreshing the connection...");
if (platform_tick_get_ms() - client->keepalive_tick > client->connect_info.keepalive * 1000 / 2) {
//No ping resp from last ping => Disconnected
if (client->wait_for_ping_resp) {
ESP_LOGE(TAG, "No PING_RESP, disconnected");
esp_mqtt_abort_connection(client);
client->state = MQTT_STATE_INIT;
client->wait_for_ping_resp = false;
break;
}
if (esp_mqtt_client_ping(client) == ESP_FAIL) {
ESP_LOGE(TAG, "Can't send ping, disconnected");
esp_mqtt_abort_connection(client);
break;
} else {
client->wait_for_ping_resp = true;
}
ESP_LOGD(TAG, "PING sent");
}
//Delete mesaage after 30 senconds
outbox_delete_expired(client->outbox, platform_tick_get_ms(), OUTBOX_EXPIRED_TIMEOUT_MS);
//
outbox_cleanup(client->outbox, OUTBOX_MAX_SIZE);
if (client->config->refresh_connection_after_ms &&
platform_tick_get_ms() - 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;
}
//Delete mesaage after 30 senconds
int deleted = outbox_delete_expired(client->outbox, platform_tick_get_ms(), OUTBOX_EXPIRED_TIMEOUT_MS);
client->mqtt_state.pending_msg_count -= deleted;
if (client->mqtt_state.pending_msg_count < 0) {
client->mqtt_state.pending_msg_count = 0;
}
//
outbox_cleanup(client->outbox, OUTBOX_MAX_SIZE);
break;
case MQTT_STATE_WAIT_TIMEOUT:
if (!client->config->auto_reconnect) {
client->run = false;
client->state = MQTT_STATE_UNKNOWN;
break;
case MQTT_STATE_WAIT_TIMEOUT:
if (!client->config->auto_reconnect) {
client->run = false;
client->state = MQTT_STATE_UNKNOWN;
break;
}
if (platform_tick_get_ms() - client->reconnect_tick > client->wait_timeout_ms) {
client->state = MQTT_STATE_INIT;
client->reconnect_tick = platform_tick_get_ms();
ESP_LOGD(TAG, "Reconnecting...");
break;
}
MQTT_API_UNLOCK(client);
xEventGroupWaitBits(client->status_bits, RECONNECT_BIT, false, true,
client->wait_timeout_ms / 2 / portTICK_RATE_MS);
// continue the while loop insted of break, as the mutex is unlocked
continue;
}
if (platform_tick_get_ms() - client->reconnect_tick > client->wait_timeout_ms) {
client->state = MQTT_STATE_INIT;
client->reconnect_tick = platform_tick_get_ms();
ESP_LOGD(TAG, "Reconnecting...");
break;
}
MQTT_API_UNLOCK(client);
xEventGroupWaitBits(client->status_bits, RECONNECT_BIT, false, true,
client->wait_timeout_ms / 2 / portTICK_RATE_MS);
// continue the while loop insted of break, as the mutex is unlocked
continue;
}
MQTT_API_UNLOCK(client);
if (MQTT_STATE_CONNECTED == client->state) {
@ -945,22 +1060,26 @@ static void esp_mqtt_task(void *pv)
esp_err_t esp_mqtt_client_start(esp_mqtt_client_handle_t client)
{
if (!client) {
ESP_LOGE(TAG, "Client was not initialized");
return ESP_ERR_INVALID_ARG;
}
if (client->state >= MQTT_STATE_INIT) {
ESP_LOGE(TAG, "Client has started");
return ESP_FAIL;
}
#if MQTT_CORE_SELECTION_ENABLED
ESP_LOGD(TAG, "Core selection enabled on %u", MQTT_TASK_CORE);
if (xTaskCreatePinnedToCore(esp_mqtt_task, "mqtt_task", client->config->task_stack, client, client->config->task_prio, &client->task_handle, MQTT_TASK_CORE) != pdTRUE) {
ESP_LOGE(TAG, "Error create mqtt task");
return ESP_FAIL;
}
ESP_LOGD(TAG, "Core selection enabled on %u", MQTT_TASK_CORE);
if (xTaskCreatePinnedToCore(esp_mqtt_task, "mqtt_task", client->config->task_stack, client, client->config->task_prio, &client->task_handle, MQTT_TASK_CORE) != pdTRUE) {
ESP_LOGE(TAG, "Error create mqtt task");
return ESP_FAIL;
}
#else
ESP_LOGD(TAG, "Core selection disabled");
if (xTaskCreate(esp_mqtt_task, "mqtt_task", client->config->task_stack, client, client->config->task_prio, &client->task_handle) != pdTRUE) {
ESP_LOGE(TAG, "Error create mqtt task");
return ESP_FAIL;
}
ESP_LOGD(TAG, "Core selection disabled");
if (xTaskCreate(esp_mqtt_task, "mqtt_task", client->config->task_stack, client, client->config->task_prio, &client->task_handle) != pdTRUE) {
ESP_LOGE(TAG, "Error create mqtt task");
return ESP_FAIL;
}
#endif
return ESP_OK;
}
@ -987,7 +1106,7 @@ esp_err_t esp_mqtt_client_stop(esp_mqtt_client_handle_t client)
ESP_LOGE(TAG, "Error sending disconnect message");
}
MQTT_API_UNLOCK_FROM_OTHER_TASK(client);
client->run = false;
xEventGroupWaitBits(client->status_bits, STOPPED_BIT, false, true, portMAX_DELAY);
client->state = MQTT_STATE_UNKNOWN;
@ -1067,7 +1186,12 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
{
uint16_t pending_msg_id = 0;
if (len <= 0) {
/* Acceptable publish messages:
data == NULL, len == 0: publish null message
data valid, len == 0: publish all data, payload len is determined from string length
data valid, len > 0: publish data with defined length
*/
if (len <= 0 && data != NULL) {
len = strlen(data);
}
@ -1077,6 +1201,11 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
qos, retain,
&pending_msg_id);
if (publish_msg == NULL) {
ESP_LOGE(TAG, "Publish message cannot be created");
MQTT_API_UNLOCK_FROM_OTHER_TASK(client);
return 0;
}
/* We have to set as pending all the qos>0 messages */
if (qos > 0) {
client->mqtt_state.outbound_message = publish_msg;
@ -1115,7 +1244,7 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
current_data += data_sent;
if (remaining_len > 0) {
mqtt_connection_t* connection = &client->mqtt_state.mqtt_connection;
mqtt_connection_t *connection = &client->mqtt_state.mqtt_connection;
ESP_LOGD(TAG, "Sending fragmented message, remains to send %d bytes of %d", remaining_len, len);
if (connection->message.fragmented_msg_data_offset) {
// asked to enqueue oversized message (first time only)
@ -1123,7 +1252,7 @@ int esp_mqtt_client_publish(esp_mqtt_client_handle_t client, const char *topic,
connection->message.fragmented_msg_total_length = 0;
if (qos > 0) {
// internally enqueue all big messages, as they dont fit 'pending msg' structure
mqtt_enqueue_oversized(client, (uint8_t*)current_data, remaining_len);
mqtt_enqueue_oversized(client, (uint8_t *)current_data, remaining_len);
}
}