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

Add error handling for mqtt, fixed some issues

Browse Source
This commit is contained in:
Tuan PM
2018-05-03 21:50:24 +07:00
parent 4bc37bb8db
commit 75d3c65cb6
13 changed files with 414 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
include/mqtt_client.h
View File

@ -54,14 +54,14 @@ typedef esp_err_t (* mqtt_event_callback_t)(esp_mqtt_event_handle_t event);
typedef struct {
mqtt_event_callback_t event_handle;
char host[MQTT_MAX_HOST_LEN];
char uri[MQTT_MAX_HOST_LEN];
const char *host;
const char *uri;
uint32_t port;
char client_id[MQTT_MAX_CLIENT_LEN];
char username[MQTT_MAX_USERNAME_LEN];
char password[MQTT_MAX_PASSWORD_LEN];
char lwt_topic[MQTT_MAX_LWT_TOPIC];
char lwt_msg[MQTT_MAX_LWT_MSG];
const char *client_id;
const char *username;
const char *password;
const char *lwt_topic;
const char *lwt_msg;
int lwt_qos;
int lwt_retain;
int lwt_msg_len;

4
include/mqtt_config.h
View File

@ -60,8 +60,8 @@
#endif
#define MQTT_ENABLE_SSL CONFIG_MQTT_TRANSPORT_SSL
#define MQTT_ENABLE_WS CONFIG_MQTT_TRANSPORT_WS
#define MQTT_ENABLE_WSS CONFIG_MQTT_TRANSPORT_WSS
#define MQTT_ENABLE_WS CONFIG_MQTT_TRANSPORT_WEBSOCKET
#define MQTT_ENABLE_WSS CONFIG_MQTT_TRANSPORT_WEBSOCKET_SECURE
#define OUTBOX_EXPIRED_TIMEOUT_MS (30*1000)
#define OUTBOX_MAX_SIZE (4*1024)

8
lib/include/platform_esp32_idf.h
View File

@ -23,9 +23,13 @@
#include "esp_log.h"
#include "esp_system.h"
#define mem_assert(x) assert(x)
char *platform_create_id_string();
int platform_random(int max);
long long platform_tick_get_ms();
void ms_to_timeval(int timeout_ms, struct timeval *tv);
#define ESP_MEM_CHECK(TAG, a, action) if (!(a)) { \
ESP_LOGE(TAG,"%s:%d (%s): %s", __FILE__, __LINE__, __FUNCTION__, "Memory exhausted"); \
action; \
}
#endif

180
lib/include/transport.h
View File

@ -6,7 +6,7 @@
#ifndef _TRANSPORT_H_
#define _TRANSPORT_H_
#include "platform.h"
#include <esp_err.h>
#ifdef __cplusplus
extern "C" {
@ -17,80 +17,220 @@ typedef struct transport_list_t* transport_list_handle_t;
typedef struct transport_item_t* transport_handle_t;
typedef int (*connect_func)(transport_handle_t t, const char *host, int port, int timeout_ms);
typedef int (*io_func)(transport_handle_t t, char *buffer, int len, int timeout_ms);
typedef int (*io_func)(transport_handle_t t, const char *buffer, int len, int timeout_ms);
typedef int (*io_read_func)(transport_handle_t t, char *buffer, int len, int timeout_ms);
typedef int (*trans_func)(transport_handle_t t);
typedef int (*poll_func)(transport_handle_t t, int timeout_ms);
/**
* @brief Create transport list
*
* @return A handle can hold all transports
*/
transport_list_handle_t transport_list_init();
esp_err_t transport_list_destroy(transport_list_handle_t head);
esp_err_t transport_list_add(transport_list_handle_t head, transport_handle_t t, const char *scheme);
esp_err_t transport_list_clean(transport_list_handle_t head);
transport_handle_t transport_list_get_transport(transport_list_handle_t head, const char *tag);
/**
* @brief Cleanup and free all transports, include itself,
* this function will invoke transport_destroy of every transport have added this the list
*
* @param[in] list The list
*
* @return
* - ESP_OK
* - ESP_FAIL
*/
esp_err_t transport_list_destroy(transport_list_handle_t list);
/**
* @brief Add a transport to the list, and define a scheme to indentify this transport in the list
*
* @param[in] list The list
* @param[in] t The Transport
* @param[in] scheme The scheme
*
* @return
* - ESP_OK
*/
esp_err_t transport_list_add(transport_list_handle_t list, transport_handle_t t, const char *scheme);
/**
* @brief This function will remove all transport from the list,
* invoke transport_destroy of every transport have added this the list
*
* @param[in] list The list
*
* @return
* - ESP_OK
* - ESP_ERR_INVALID_ARG
*/
esp_err_t transport_list_clean(transport_list_handle_t list);
/**
* @brief Get the transport by scheme, which has been defined when calling function `transport_list_add`
*
* @param[in] list The list
* @param[in] tag The tag
*
* @return The transport handle
*/
transport_handle_t transport_list_get_transport(transport_list_handle_t list, const char *scheme);
/**
* @brief Initialize a transport handle object
*
* @return The transport handle
*/
transport_handle_t transport_init();
int transport_destroy(transport_handle_t t);
/**
* @brief Cleanup and free memory the transport
*
* @param[in] t The transport handle
*
* @return
* - ESP_OK
* - ESP_FAIL
*/
esp_err_t transport_destroy(transport_handle_t t);
/**
* @brief Get default port number used by this transport
*
* @param[in] t The transport handle
*
* @return the port number
*/
int transport_get_default_port(transport_handle_t t);
/**
* @brief Set default port number that can be used by this transport
*
* @param[in] t The transport handle
* @param[in] port The port number
*
* @return
* - ESP_OK
* - ESP_FAIL
*/
esp_err_t transport_set_default_port(transport_handle_t t, int port);
/**
* @brief Transport connection function, to make a connection to server
*
* @param t Transport to use
* @param t The transport handle
* @param[in] host Hostname
* @param[in] port Port
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - socket for will use by this transport
* - (-1) if there are any errors
* - (-1) if there are any errors, should check errno
*/
int transport_connect(transport_handle_t t, const char *host, int port, int timeout_ms);
/**
* @brief Transport read function
*
* @param t Transport to use
* @param t The transport handle
* @param buffer The buffer
* @param[in] len The length
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - Number of bytes was read
* - (-1) if there are any errors
* - (-1) if there are any errors, should check errno
*/
int transport_read(transport_handle_t t, char *buffer, int len, int timeout_ms);
/**
* @brief Poll the transport until readable or timeout
*
* @param[in] t The transport handle
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - 0 Timeout
* - (-1) If there are any errors, should check errno
* - other The transport can read
*/
int transport_poll_read(transport_handle_t t, int timeout_ms);
/**
* @brief Transport write function
*
* @param t transport
* @param t The transport handle
* @param buffer The buffer
* @param[in] len The length
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - Number of bytes was written
* - (-1) if there are any errors
* - Number of bytes was written
* - (-1) if there are any errors, should check errno
*/
int transport_write(transport_handle_t t, const char *buffer, int len, int timeout_ms);
/**
* @brief Poll the transport until writeable or timeout
*
* @param[in] t The transport handle
* @param[in] timeout_ms The timeout milliseconds
*
* @return
* - 0 Timeout
* - (-1) If there are any errors, should check errno
* - other The transport can write
*/
int transport_write(transport_handle_t t, char *buffer, int len, int timeout_ms);
int transport_poll_write(transport_handle_t t, int timeout_ms);
/**
* @brief Transport close
*
* @param t transport
* @param t The transport handle
*
* @return
* - 0 if ok
* - (-1) if there are any errors
* - (-1) if there are any errors, should check errno
*/
int transport_close(transport_handle_t t);
void *transport_get_data(transport_handle_t t);
esp_err_t transport_set_data(transport_handle_t t, void *data);
/**
* @brief Get user data context of this transport
*
* @param[in] t The transport handle
*
* @return The user data context
*/
void *transport_get_context_data(transport_handle_t t);
/**
* @brief Set the user context data for this transport
*
* @param[in] t The transport handle
* @param data The user data context
*
* @return
* - ESP_OK
*/
esp_err_t transport_set_context_data(transport_handle_t t, void *data);
/**
* @brief Set transport functions for the transport handle
*
* @param[in] t The transport handle
* @param[in] _connect The connect function pointer
* @param[in] _read The read function pointer
* @param[in] _write The write function pointer
* @param[in] _close The close function pointer
* @param[in] _poll_read The poll read function pointer
* @param[in] _poll_write The poll write function pointer
* @param[in] _destroy The destroy function pointer
*
* @return
* - ESP_OK
*/
esp_err_t transport_set_func(transport_handle_t t,
connect_func _connect,
io_func _read,
io_read_func _read,
io_func _write,
trans_func _close,
poll_func _poll_read,

11
lib/include/transport_ssl.h
View File

@ -14,16 +14,16 @@ extern "C" {
/**
* @brief Create new SSL transport
* @brief Create new SSL transport, the transport handle must be release transport_destroy callback
*
* @return
* - transport
* - NULL
* @return the allocated transport_handle_t, or NULL if the handle can not be allocated
*/
transport_handle_t transport_ssl_init();
/**
* @brief Set SSL certification data (as PEM format)
* @brief Set SSL certificate data (as PEM format).
* Note that, this function stores the pointer to data, rather than making a copy.
* So we need to make sure to keep the data lifetime before cleanup the connection
*
* @param t ssl transport
* @param[in] data The pem data
@ -36,3 +36,4 @@ void transport_ssl_set_cert_data(transport_handle_t t, const char *data, int len
}
#endif
#endif

6
lib/include/transport_tcp.h
View File

@ -13,11 +13,9 @@ extern "C" {
#endif
/**
* @brief Create TCP transport
* @brief Create TCP transport, the transport handle must be release transport_destroy callback
*
* @return
* - transport
* - NULL
* @return the allocated transport_handle_t, or NULL if the handle can not be allocated
*/
transport_handle_t transport_tcp_init();

9
lib/mqtt_outbox.c
View File

@ -9,7 +9,7 @@ static const char *TAG = "OUTBOX";
outbox_handle_t outbox_init()
{
outbox_handle_t outbox = calloc(1, sizeof(struct outbox_list_t));
mem_assert(outbox);
ESP_MEM_CHECK(TAG, outbox, return NULL);
STAILQ_INIT(outbox);
return outbox;
}
@ -17,13 +17,16 @@ outbox_handle_t outbox_init()
outbox_item_handle_t outbox_enqueue(outbox_handle_t outbox, uint8_t *data, int len, int msg_id, int msg_type, int tick)
{
outbox_item_handle_t item = calloc(1, sizeof(outbox_item_t));
mem_assert(item);
ESP_MEM_CHECK(TAG, item, return NULL);
item->msg_id = msg_id;
item->msg_type = msg_type;
item->tick = tick;
item->len = len;
item->buffer = malloc(len);
mem_assert(item->buffer);
ESP_MEM_CHECK(TAG, item->buffer, {
free(item);
return NULL;
});
memcpy(item->buffer, data, len);
STAILQ_INSERT_TAIL(outbox, item, next);
ESP_LOGD(TAG, "ENQUEUE msgid=%d, msg_type=%d, len=%d, size=%d", msg_id, msg_type, len, outbox_get_size(outbox));

15
lib/platform_esp32_idf.c
View File

@ -2,13 +2,18 @@
#ifdef ESP_PLATFORM
#include "esp_system.h"
#include "esp_log.h"
#include <sys/time.h>
static const char *TAG = "PLATFORM";
#define MAX_ID_STRING (32)
char *platform_create_id_string()
{
uint8_t mac[6];
char *id_string = calloc(1, 32);
mem_assert(id_string);
char *id_string = calloc(1, MAX_ID_STRING);
ESP_MEM_CHECK(TAG, id_string, return NULL);
esp_read_mac(mac, ESP_MAC_WIFI_STA);
sprintf(id_string, "ESP32_%02x%02X%02X", mac[3], mac[4], mac[5]);
return id_string;
@ -27,4 +32,10 @@ long long platform_tick_get_ms()
// printf("milliseconds: %lld\n", milliseconds);
return milliseconds;
}
void ms_to_timeval(int timeout_ms, struct timeval *tv)
{
tv->tv_sec = timeout_ms / 1000;
tv->tv_usec = (timeout_ms - (tv->tv_sec * 1000)) * 1000;
}
#endif

100
lib/transport.c
View File

@ -1,25 +1,31 @@
#include <stdlib.h>
#include <string.h>
#include "esp_log.h"
#include "transport.h"
// static const char *TAG = "TRANSPORT";
#include "rom/queue.h"
#include "esp_log.h"
#include "transport.h"
#include "platform.h"
static const char *TAG = "TRANSPORT";
/**
* Transport layer structure, which will provide functions, basic properties for transport types
*/
struct transport_item_t {
int port;
int socket; /*!< Socket to use in this transport */
char *tag; /*!< Tag name */
void *context; /*!< Context data */
void *data; /*!< Additional transport data */
connect_func _connect; /*!< Connect function of this transport */
io_func _read; /*!< Read */
io_func _write; /*!< Write */
trans_func _close; /*!< Close */
poll_func _poll_read; /*!< Poll and read */
poll_func _poll_write; /*!< Poll and write */
trans_func _destroy; /*!< Destroy and free transport */
int port;
int socket; /*!< Socket to use in this transport */
char *scheme; /*!< Tag name */
void *context; /*!< Context data */
void *data; /*!< Additional transport data */
connect_func _connect; /*!< Connect function of this transport */
io_read_func _read; /*!< Read */
io_func _write; /*!< Write */
trans_func _close; /*!< Close */
poll_func _poll_read; /*!< Poll and read */
poll_func _poll_write; /*!< Poll and write */
trans_func _destroy; /*!< Destroy and free transport */
STAILQ_ENTRY(transport_item_t) next;
};
@ -32,52 +38,51 @@ STAILQ_HEAD(transport_list_t, transport_item_t);
transport_list_handle_t transport_list_init()
{
transport_list_handle_t head = calloc(1, sizeof(struct transport_list_t));
assert(head);
STAILQ_INIT(head);
return head;
transport_list_handle_t list = calloc(1, sizeof(struct transport_list_t));
ESP_MEM_CHECK(TAG, list, return NULL);
STAILQ_INIT(list);
return list;
}
esp_err_t transport_list_add(transport_list_handle_t head, transport_handle_t t, const char *scheme)
esp_err_t transport_list_add(transport_list_handle_t list, transport_handle_t t, const char *scheme)
{
if (head == NULL || t == NULL) {
if (list == NULL || t == NULL) {
return ESP_ERR_INVALID_ARG;
}
t->tag = calloc(1, strlen(scheme) + 1);
assert(t->tag);
strcpy(t->tag, scheme);
STAILQ_INSERT_TAIL(head, t, next);
t->scheme = calloc(1, strlen(scheme) + 1);
ESP_MEM_CHECK(TAG, t->scheme, return ESP_ERR_NO_MEM);
strcpy(t->scheme, scheme);
STAILQ_INSERT_TAIL(list, t, next);
return ESP_OK;
}
transport_handle_t transport_list_get_transport(transport_list_handle_t head, const char *tag)
transport_handle_t transport_list_get_transport(transport_list_handle_t list, const char *scheme)
{
if (!head) {
if (!list) {
return NULL;
}
if (tag == NULL) {
return STAILQ_FIRST(head);
if (scheme == NULL) {
return STAILQ_FIRST(list);
}
transport_handle_t item;
STAILQ_FOREACH(item, head, next) {
if (strcasecmp(item->tag, tag) == 0) {
STAILQ_FOREACH(item, list, next) {
if (strcasecmp(item->scheme, scheme) == 0) {
return item;
}
}
return NULL;
}
esp_err_t transport_list_destroy(transport_list_handle_t head)
esp_err_t transport_list_destroy(transport_list_handle_t list)
{
transport_list_clean(head);
free(head);
transport_list_clean(list);
free(list);
return ESP_OK;
}
esp_err_t transport_list_clean(transport_list_handle_t head)
esp_err_t transport_list_clean(transport_list_handle_t list)
{
transport_handle_t item = STAILQ_FIRST(head);
transport_handle_t item = STAILQ_FIRST(list);
transport_handle_t tmp;
while (item != NULL) {
tmp = STAILQ_NEXT(item, next);
@ -87,24 +92,24 @@ esp_err_t transport_list_clean(transport_list_handle_t head)
transport_destroy(item);
item = tmp;
}
STAILQ_INIT(head);
STAILQ_INIT(list);
return ESP_OK;
}
transport_handle_t transport_init()
{
transport_handle_t t = calloc(1, sizeof(struct transport_item_t));
assert(t);
ESP_MEM_CHECK(TAG, t, return NULL);
return t;
}
int transport_destroy(transport_handle_t t)
esp_err_t transport_destroy(transport_handle_t t)
{
if (t->tag) {
free(t->tag);
if (t->scheme) {
free(t->scheme);
}
free(t);
return 0;
return ESP_OK;
}
int transport_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
@ -124,7 +129,7 @@ int transport_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
return -1;
}
int transport_write(transport_handle_t t, char *buffer, int len, int timeout_ms)
int transport_write(transport_handle_t t, const char *buffer, int len, int timeout_ms)
{
if (t && t->_write) {
return t->_write(t, buffer, len, timeout_ms);
@ -156,7 +161,7 @@ int transport_close(transport_handle_t t)
return 0;
}
void *transport_get_data(transport_handle_t t)
void *transport_get_context_data(transport_handle_t t)
{
if (t) {
return t->data;
@ -164,7 +169,7 @@ void *transport_get_data(transport_handle_t t)
return NULL;
}
esp_err_t transport_set_data(transport_handle_t t, void *data)
esp_err_t transport_set_context_data(transport_handle_t t, void *data)
{
if (t) {
t->data = data;
@ -175,7 +180,7 @@ esp_err_t transport_set_data(transport_handle_t t, void *data)
esp_err_t transport_set_func(transport_handle_t t,
connect_func _connect,
io_func _read,
io_read_func _read,
io_func _write,
trans_func _close,
poll_func _poll_read,
@ -211,4 +216,3 @@ esp_err_t transport_set_default_port(transport_handle_t t, int port)
t->port = port;
return ESP_OK;
}

83
lib/transport_ssl.c
View File

@ -1,6 +1,14 @@
#include <string.h>
#include <stdlib.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "lwip/err.h"
#include "lwip/sockets.h"
#include "lwip/sys.h"
#include "lwip/netdb.h"
#include "lwip/dns.h"
#include "mbedtls/platform.h"
#include "mbedtls/net_sockets.h"
#include "mbedtls/esp_debug.h"
@ -13,12 +21,12 @@
#include "esp_log.h"
#include "esp_system.h"
#include "platform.h"
#include "transport.h"
#include "transport_ssl.h"
static const char *TAG = "TRANSPORT_SSL";
static const char *TAG = "TRANS_SSL";
/**
* mbedtls specific transport data
*/
@ -35,11 +43,13 @@ typedef struct {
bool verify_server;
} transport_ssl_t;
static int ssl_close(transport_handle_t t);
static int ssl_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
{
int ret = -1, flags;
struct timeval tv;
transport_ssl_t *ssl = transport_get_data(t);
transport_ssl_t *ssl = transport_get_context_data(t);
if (!ssl) {
return -1;
@ -95,12 +105,8 @@ static int ssl_connect(transport_handle_t t, const char *host, int port, int tim
mbedtls_net_init(&ssl->client_fd);
tv.tv_sec = 10; //default timeout is 10 seconds
ms_to_timeval(timeout_ms, &tv);
if (timeout_ms) {
tv.tv_sec = timeout_ms;
}
tv.tv_usec = 0;
setsockopt(ssl->client_fd.fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
ESP_LOGD(TAG, "Connect to %s:%d", host, port);
char port_str[8] = {0};
@ -112,6 +118,11 @@ static int ssl_connect(transport_handle_t t, const char *host, int port, int tim
mbedtls_ssl_set_bio(&ssl->ctx, &ssl->client_fd, mbedtls_net_send, mbedtls_net_recv, NULL);
if((ret = mbedtls_ssl_set_hostname(&ssl->ctx, host)) != 0) {
ESP_LOGE(TAG, " failed\n ! mbedtls_ssl_set_hostname returned %d\n\n", ret);
goto exit;
}
ESP_LOGD(TAG, "Performing the SSL/TLS handshake...");
while ((ret = mbedtls_ssl_handshake(&ssl->ctx)) != 0) {
@ -127,68 +138,62 @@ static int ssl_connect(transport_handle_t t, const char *host, int port, int tim
/* In real life, we probably want to close connection if ret != 0 */
ESP_LOGW(TAG, "Failed to verify peer certificate!");
if (ssl->cert_pem_data) {
return -1;
goto exit;
}
// bzero(buf, sizeof(buf));
// mbedtls_x509_crt_verify_info(buf, sizeof(buf), " ! ", flags);
// ESP_LOGW(TAG, "verification info: %s", buf);
} else {
ESP_LOGD(TAG, "Certificate verified.");
}
ESP_LOGD(TAG, "Cipher suite is %s", mbedtls_ssl_get_ciphersuite(&ssl->ctx));
return ret;
exit:
ssl_close(t);
return ret;
}
static int ssl_poll_read(transport_handle_t t, int timeout_ms)
{
transport_ssl_t *ssl = transport_get_data(t);
transport_ssl_t *ssl = transport_get_context_data(t);
fd_set readset;
FD_ZERO(&readset);
FD_SET(ssl->client_fd.fd, &readset);
struct timeval timeout;
timeout.tv_sec = timeout_ms / 1000;
timeout.tv_usec = (timeout_ms % 1000) * 1000;
ms_to_timeval(timeout_ms, &timeout);
return select(ssl->client_fd.fd + 1, &readset, NULL, NULL, &timeout);
}
static int ssl_poll_write(transport_handle_t t, int timeout_ms)
{
transport_ssl_t *ssl = transport_get_data(t);
transport_ssl_t *ssl = transport_get_context_data(t);
fd_set writeset;
FD_ZERO(&writeset);
FD_SET(ssl->client_fd.fd, &writeset);
struct timeval timeout;
timeout.tv_sec = timeout_ms / 1000;
timeout.tv_usec = (timeout_ms % 1000) * 1000;
ms_to_timeval(timeout_ms, &timeout);
return select(ssl->client_fd.fd + 1, NULL, &writeset, NULL, &timeout);
}
static int ssl_write(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int ssl_write(transport_handle_t t, const char *buffer, int len, int timeout_ms)
{
int poll, ret;
transport_ssl_t *ssl = transport_get_data(t);
transport_ssl_t *ssl = transport_get_context_data(t);
if ((poll = transport_poll_write(t, timeout_ms)) <= 0) {
ESP_LOGW(TAG, "Poll timeout or error, errno=%s, fd=%d, timeout_ms=%d", strerror(errno), ssl->client_fd.fd, timeout_ms);
return poll;
}
ret = mbedtls_ssl_write(&ssl->ctx, (const unsigned char *) buffer, len);
//TODO: Debug here
if (ret <= 0) {
ESP_LOGE(TAG, "mbedtls_ssl_write error, errno=%s", strerror(errno));
}
return ret;
}
static int ssl_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
int poll = -1, ret;
transport_ssl_t *ssl = transport_get_data(t);
if ((poll = transport_poll_read(t, timeout_ms)) <= 0) {
return poll;
}
int ret;
transport_ssl_t *ssl = transport_get_context_data(t);
ret = mbedtls_ssl_read(&ssl->ctx, (unsigned char *)buffer, len);
if (ret == 0) {
return -1;
@ -199,7 +204,7 @@ static int ssl_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int ssl_close(transport_handle_t t)
{
int ret = -1;
transport_ssl_t *ssl = transport_get_data(t);
transport_ssl_t *ssl = transport_get_context_data(t);
if (ssl->ssl_initialized) {
ESP_LOGD(TAG, "Cleanup mbedtls");
mbedtls_ssl_close_notify(&ssl->ctx);
@ -218,10 +223,9 @@ static int ssl_close(transport_handle_t t)
return ret;
}
static int ssl_destroy(transport_handle_t t)
{
transport_ssl_t *ssl = transport_get_data(t);
transport_ssl_t *ssl = transport_get_context_data(t);
transport_close(t);
free(ssl);
return 0;
@ -229,12 +233,10 @@ static int ssl_destroy(transport_handle_t t)
void transport_ssl_set_cert_data(transport_handle_t t, const char *data, int len)
{
transport_ssl_t *ssl = transport_get_data(t);
if (t) {
if (t && ssl) {
ssl->cert_pem_data = (void *)data;
ssl->cert_pem_len = len;
}
transport_ssl_t *ssl = transport_get_context_data(t);
if (t && ssl) {
ssl->cert_pem_data = (void *)data;
ssl->cert_pem_len = len;
}
}
@ -242,9 +244,10 @@ transport_handle_t transport_ssl_init()
{
transport_handle_t t = transport_init();
transport_ssl_t *ssl = calloc(1, sizeof(transport_ssl_t));
assert(ssl);
ESP_MEM_CHECK(TAG, ssl, return NULL);
mbedtls_net_init(&ssl->client_fd);
transport_set_data(t, ssl);
transport_set_context_data(t, ssl);
transport_set_func(t, ssl_connect, ssl_read, ssl_write, ssl_close, ssl_poll_read, ssl_poll_write, ssl_destroy);
return t;
}

44
lib/transport_tcp.c
View File

@ -1,16 +1,25 @@
#include <stdlib.h>
#include <string.h>
#include "lwip/sockets.h"
#include "lwip/dns.h"
#include "lwip/netdb.h"
#include "esp_log.h"
#include "esp_system.h"
#include "esp_err.h"
#include "platform.h"
#include "transport.h"
static const char *TAG = "TRANSPORT_TCP";
static const char *TAG = "TRANS_TCP";
typedef struct {
int sock;
} transport_tcp_t;
static int resolve_dns(const char *host, struct sockaddr_in *ip) {
struct hostent *he;
struct in_addr **addr_list;
he = gethostbyname(host);
@ -30,7 +39,7 @@ static int tcp_connect(transport_handle_t t, const char *host, int port, int tim
{
struct sockaddr_in remote_ip;
struct timeval tv;
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
bzero(&remote_ip, sizeof(struct sockaddr_in));
@ -51,12 +60,8 @@ static int tcp_connect(transport_handle_t t, const char *host, int port, int tim
remote_ip.sin_family = AF_INET;
remote_ip.sin_port = htons(port);
tv.tv_sec = 10; //default timeout is 10 seconds
ms_to_timeval(timeout_ms, &tv);
if (timeout_ms) {
tv.tv_sec = timeout_ms;
}
tv.tv_usec = 0;
setsockopt(tcp->sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
ESP_LOGD(TAG, "[sock=%d],connecting to server IP:%s,Port:%d...",
@ -69,10 +74,10 @@ static int tcp_connect(transport_handle_t t, const char *host, int port, int tim
return tcp->sock;
}
static int tcp_write(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int tcp_write(transport_handle_t t, const char *buffer, int len, int timeout_ms)
{
int poll;
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
if ((poll = transport_poll_write(t, timeout_ms)) <= 0) {
return poll;
}
@ -81,7 +86,7 @@ static int tcp_write(transport_handle_t t, char *buffer, int len, int timeout_ms
static int tcp_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
int poll = -1;
if ((poll = transport_poll_read(t, timeout_ms)) <= 0) {
return poll;
@ -95,31 +100,29 @@ static int tcp_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int tcp_poll_read(transport_handle_t t, int timeout_ms)
{
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
fd_set readset;
FD_ZERO(&readset);
FD_SET(tcp->sock, &readset);
struct timeval timeout;
timeout.tv_sec = timeout_ms / 1000;
timeout.tv_usec = (timeout_ms % 1000) * 1000;
ms_to_timeval(timeout_ms, &timeout);
return select(tcp->sock + 1, &readset, NULL, NULL, &timeout);
}
static int tcp_poll_write(transport_handle_t t, int timeout_ms)
{
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
fd_set writeset;
FD_ZERO(&writeset);
FD_SET(tcp->sock, &writeset);
struct timeval timeout;
timeout.tv_sec = timeout_ms / 1000;
timeout.tv_usec = (timeout_ms % 1000) * 1000;
ms_to_timeval(timeout_ms, &timeout);
return select(tcp->sock + 1, NULL, &writeset, NULL, &timeout);
}
static int tcp_close(transport_handle_t t)
{
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
int ret = -1;
if (tcp->sock >= 0) {
ret = close(tcp->sock);
@ -130,7 +133,7 @@ static int tcp_close(transport_handle_t t)
static esp_err_t tcp_destroy(transport_handle_t t)
{
transport_tcp_t *tcp = transport_get_data(t);
transport_tcp_t *tcp = transport_get_context_data(t);
transport_close(t);
free(tcp);
return 0;
@ -140,11 +143,10 @@ transport_handle_t transport_tcp_init()
{
transport_handle_t t = transport_init();
transport_tcp_t *tcp = calloc(1, sizeof(transport_tcp_t));
assert(tcp);
ESP_MEM_CHECK(TAG, tcp, return NULL);
tcp->sock = -1;
transport_set_func(t, tcp_connect, tcp_read, tcp_write, tcp_close, tcp_poll_read, tcp_poll_write, tcp_destroy);
transport_set_data(t, tcp);
transport_set_context_data(t, tcp);
return t;
}

38
lib/transport_ws.c
View File

@ -58,7 +58,7 @@ static char *get_http_header(const char *buffer, const char *key)
static int ws_connect(transport_handle_t t, const char *host, int port, int timeout_ms)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
if (transport_connect(ws->parent, host, port, timeout_ms) < 0) {
ESP_LOGE(TAG, "Error connect to ther server");
}
@ -109,13 +109,13 @@ static int ws_connect(transport_handle_t t, const char *host, int port, int time
return 0;
}
static int ws_write(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int ws_write(transport_handle_t t, const char *buff, int len, int timeout_ms)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
char ws_header[MAX_WEBSOCKET_HEADER_SIZE];
char *mask;
int header_len = 0, i;
char *buffer = (char *)buff;
int poll_write;
if ((poll_write = transport_poll_write(ws->parent, timeout_ms)) <= 0) {
return poll_write;
@ -149,7 +149,7 @@ static int ws_write(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int ws_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
int payload_len;
char *data_ptr = buffer, opcode, mask, *mask_key = NULL;
int rlen;
@ -198,25 +198,25 @@ static int ws_read(transport_handle_t t, char *buffer, int len, int timeout_ms)
static int ws_poll_read(transport_handle_t t, int timeout_ms)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
return transport_poll_read(ws->parent, timeout_ms);
}
static int ws_poll_write(transport_handle_t t, int timeout_ms)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
return transport_poll_write(ws->parent, timeout_ms);;
}
static int ws_close(transport_handle_t t)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
return transport_close(ws->parent);
}
static esp_err_t ws_destroy(transport_handle_t t)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
free(ws->buffer);
free(ws->path);
free(ws);
@ -224,7 +224,7 @@ static esp_err_t ws_destroy(transport_handle_t t)
}
void transport_ws_set_path(transport_handle_t t, const char *path)
{
transport_ws_t *ws = transport_get_data(t);
transport_ws_t *ws = transport_get_context_data(t);
ws->path = realloc(ws->path, strlen(path) + 1);
strcpy(ws->path, path);
}
@ -232,16 +232,20 @@ transport_handle_t transport_ws_init(transport_handle_t parent_handle)
{
transport_handle_t t = transport_init();
transport_ws_t *ws = calloc(1, sizeof(transport_ws_t));
assert(ws);
ESP_MEM_CHECK(TAG, ws, return NULL);
ws->parent = parent_handle;
ws->buffer = malloc(DEFAULT_WS_BUFFER);
mem_assert(ws->buffer);
ws->path = calloc(1, 2);
ws->path[0] = '/';
mem_assert(ws->buffer);
ws->path = strdup("/");
ESP_MEM_CHECK(TAG, ws->path, return NULL);
ws->buffer = malloc(DEFAULT_WS_BUFFER);
ESP_MEM_CHECK(TAG, ws->buffer, {
free(ws->path);
free(ws);
return NULL;
});
transport_set_func(t, ws_connect, ws_read, ws_write, ws_close, ws_poll_read, ws_poll_write, ws_destroy);
transport_set_data(t, ws);
transport_set_context_data(t, ws);
return t;
}

136
mqtt_client.c
View File

@ -84,8 +84,11 @@ static char *create_string(const char *ptr, int len);
static esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config)
{
//Copy user configurations to client context
esp_err_t err = ESP_OK;
mqtt_config_storage_t *cfg = calloc(1, sizeof(mqtt_config_storage_t));
mem_assert(cfg);
ESP_MEM_CHECK(TAG, cfg, return ESP_ERR_NO_MEM);
client->config = cfg;
cfg->task_prio = config->task_prio;
if (cfg->task_prio <= 0) {
@ -96,42 +99,49 @@ static esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_
if (cfg->task_stack == 0) {
cfg->task_stack = MQTT_TASK_STACK;
}
if (config->host[0]) {
err = ESP_ERR_NO_MEM;
if (config->host) {
cfg->host = strdup(config->host);
ESP_MEM_CHECK(TAG, cfg->host, goto _mqtt_set_config_failed);
}
cfg->port = config->port;
if (config->username[0]) {
if (config->username) {
client->connect_info.username = strdup(config->username);
ESP_MEM_CHECK(TAG, client->connect_info.username, goto _mqtt_set_config_failed);
}
if (config->password[0]) {
if (config->password) {
client->connect_info.password = strdup(config->password);
ESP_MEM_CHECK(TAG, client->connect_info.password, goto _mqtt_set_config_failed);
}
if (config->client_id[0]) {
if (config->client_id) {
client->connect_info.client_id = strdup(config->client_id);
} else {
client->connect_info.client_id = platform_create_id_string();
}
ESP_MEM_CHECK(TAG, client->connect_info.client_id, goto _mqtt_set_config_failed);
ESP_LOGD(TAG, "MQTT client_id=%s", client->connect_info.client_id);
if (config->uri[0]) {
if (config->uri) {
cfg->uri = strdup(config->uri);
ESP_MEM_CHECK(TAG, cfg->uri, goto _mqtt_set_config_failed);
}
if (config->lwt_topic[0]) {
if (config->lwt_topic) {
client->connect_info.will_topic = strdup(config->lwt_topic);
ESP_MEM_CHECK(TAG, client->connect_info.will_topic, goto _mqtt_set_config_failed);
}
if (config->lwt_msg_len) {
client->connect_info.will_message = malloc(config->lwt_msg_len);
mem_assert(client->connect_info.will_message);
ESP_MEM_CHECK(TAG, client->connect_info.will_message, goto _mqtt_set_config_failed);
memcpy(client->connect_info.will_message, config->lwt_msg, config->lwt_msg_len);
client->connect_info.will_length = config->lwt_msg_len;
} else if (config->lwt_msg[0]) {
} else if (config->lwt_msg) {
client->connect_info.will_message = strdup(config->lwt_msg);
ESP_MEM_CHECK(TAG, client->connect_info.will_message, goto _mqtt_set_config_failed);
client->connect_info.will_length = strlen(config->lwt_msg);
}
@ -154,40 +164,25 @@ static esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_
cfg->auto_reconnect = false;
}
client->config = cfg;
return ESP_OK;
return err;
_mqtt_set_config_failed:
esp_mqtt_destroy_config(client);
return err;
}
static esp_err_t esp_mqtt_destroy_config(esp_mqtt_client_handle_t client)
{
mqtt_config_storage_t *cfg = client->config;
if (cfg->host) {
free(cfg->host);
}
if (cfg->uri) {
free(cfg->uri);
}
if (cfg->path) {
free(cfg->path);
}
if (cfg->scheme) {
free(cfg->scheme);
}
if (client->connect_info.will_topic) {
free(client->connect_info.will_topic);
}
if (client->connect_info.will_message) {
free(client->connect_info.will_message);
}
if (client->connect_info.client_id) {
free(client->connect_info.client_id);
}
if (client->connect_info.username) {
free(client->connect_info.username);
}
if (client->connect_info.password) {
free(client->connect_info.password);
}
free(cfg->host);
free(cfg->uri);
free(cfg->path);
free(cfg->scheme);
free(client->connect_info.will_topic);
free(client->connect_info.will_message);
free(client->connect_info.client_id);
free(client->connect_info.username);
free(client->connect_info.password);
free(client->config);
return ESP_OK;
}
@ -267,30 +262,36 @@ static esp_err_t esp_mqtt_abort_connection(esp_mqtt_client_handle_t client)
esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config)
{
esp_mqtt_client_handle_t client = calloc(1, sizeof(struct esp_mqtt_client));
mem_assert(client);
ESP_MEM_CHECK(TAG, client, return NULL);
esp_mqtt_set_config(client, config);
client->transport_list = transport_list_init();
ESP_MEM_CHECK(TAG, client->transport_list, goto _mqtt_init_failed);
transport_handle_t tcp = transport_tcp_init();
ESP_MEM_CHECK(TAG, tcp, goto _mqtt_init_failed);
transport_set_default_port(tcp, MQTT_TCP_DEFAULT_PORT);
transport_list_add(client->transport_list, tcp, "mqtt");
if (config->transport == MQTT_TRANSPORT_OVER_TCP) {
client->config->scheme = create_string("mqtt", 4);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#if MQTT_ENABLE_WS
transport_handle_t ws = transport_ws_init(tcp);
ESP_MEM_CHECK(TAG, ws, goto _mqtt_init_failed);
transport_set_default_port(ws, MQTT_WS_DEFAULT_PORT);
transport_list_add(client->transport_list, ws, "ws");
if (config->transport == MQTT_TRANSPORT_OVER_WS) {
client->config->scheme = create_string("ws", 2);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#endif
#if MQTT_ENABLE_SSL
transport_handle_t ssl = transport_ssl_init();
ESP_MEM_CHECK(TAG, ssl, goto _mqtt_init_failed);
transport_set_default_port(ssl, MQTT_SSL_DEFAULT_PORT);
if (config->cert_pem) {
transport_ssl_set_cert_data(ssl, config->cert_pem, strlen(config->cert_pem));
@ -298,25 +299,29 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
transport_list_add(client->transport_list, ssl, "mqtts");
if (config->transport == MQTT_TRANSPORT_OVER_SSL) {
client->config->scheme = create_string("mqtts", 5);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#endif
#if MQTT_ENABLE_WSS
transport_handle_t wss = transport_ws_init(ssl);
ESP_MEM_CHECK(TAG, wss, goto _mqtt_init_failed);
transport_set_default_port(wss, MQTT_WSS_DEFAULT_PORT);
transport_list_add(client->transport_list, wss, "wss");
if (config->transport == MQTT_TRANSPORT_OVER_WSS) {
client->config->scheme = create_string("wss", 3);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
#endif
if (client->config->uri) {
if (esp_mqtt_client_set_uri(client, client->config->uri) != ESP_OK) {
return NULL;
goto _mqtt_init_failed;
}
}
if (client->config->scheme == NULL) {
client->config->scheme = create_string("mqtt", 4);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_init_failed);
}
client->keepalive_tick = platform_tick_get_ms();
@ -328,15 +333,21 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
}
client->mqtt_state.in_buffer = (uint8_t *)malloc(buffer_size);
mem_assert(client->mqtt_state.in_buffer);
ESP_MEM_CHECK(TAG, client->mqtt_state.in_buffer, goto _mqtt_init_failed);
client->mqtt_state.in_buffer_length = buffer_size;
client->mqtt_state.out_buffer = (uint8_t *)malloc(buffer_size);
mem_assert(client->mqtt_state.out_buffer);
ESP_MEM_CHECK(TAG, client->mqtt_state.out_buffer, goto _mqtt_init_failed);
client->mqtt_state.out_buffer_length = buffer_size;
client->mqtt_state.connect_info = &client->connect_info;
client->outbox = outbox_init();
ESP_MEM_CHECK(TAG, client->outbox, goto _mqtt_init_failed);
client->status_bits = xEventGroupCreate();
ESP_MEM_CHECK(TAG, client->status_bits, goto _mqtt_init_failed);
return client;
_mqtt_init_failed:
esp_mqtt_client_destroy(client);
return NULL;
}
esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client)
@ -359,7 +370,7 @@ static char *create_string(const char *ptr, int len)
return NULL;
}
ret = calloc(1, len + 1);
mem_assert(ret);
ESP_MEM_CHECK(TAG, ret, return NULL);
memcpy(ret, ptr, len);
return ret;
}
@ -396,10 +407,8 @@ esp_err_t esp_mqtt_client_set_uri(esp_mqtt_client_handle_t client, const char *u
}
}
char *port = create_string(uri + puri.field_data[UF_PORT].off, puri.field_data[UF_PORT].len);
if (port) {
client->config->port = atoi(port);
free(port);
if (puri.field_data[UF_PORT].len) {
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);
@ -442,7 +451,7 @@ static esp_err_t esp_mqtt_dispatch_event(esp_mqtt_client_handle_t client)
return client->config->event_handle(&client->event);
}
return ESP_FAIL;
}
}
@ -455,7 +464,7 @@ static void deliver_publish(esp_mqtt_client_handle_t client, uint8_t *message, i
do
{
if (total_mqtt_len == 0){
if (total_mqtt_len == 0) {
mqtt_topic_length = length;
mqtt_topic = mqtt_get_publish_topic(message, &mqtt_topic_length);
mqtt_data_length = length;
@ -478,14 +487,15 @@ static void deliver_publish(esp_mqtt_client_handle_t client, uint8_t *message, i
esp_mqtt_dispatch_event(client);
mqtt_offset += mqtt_len;
if (client->mqtt_state.message_length_read >= client->mqtt_state.message_length)
if (client->mqtt_state.message_length_read >= client->mqtt_state.message_length) {
break;
}
len_read = transport_read(client->transport,
(char *)client->mqtt_state.in_buffer,
client->mqtt_state.message_length - client->mqtt_state.message_length_read > client->mqtt_state.in_buffer_length ?
client->mqtt_state.in_buffer_length : client->mqtt_state.message_length - client->mqtt_state.message_length_read,
client->config->network_timeout_ms);
(char *)client->mqtt_state.in_buffer,
client->mqtt_state.message_length - client->mqtt_state.message_length_read > client->mqtt_state.in_buffer_length ?
client->mqtt_state.in_buffer_length : client->mqtt_state.message_length - client->mqtt_state.message_length_read,
client->config->network_timeout_ms);
if (len_read <= 0) {
ESP_LOGE(TAG, "Read error or timeout: %d", errno);
break;
@ -517,7 +527,7 @@ static bool is_valid_mqtt_msg(esp_mqtt_client_handle_t client, int msg_type, int
static void mqtt_enqueue(esp_mqtt_client_handle_t client)
{
ESP_LOGD(TAG, "mqtt_enqueue id: %d, type=%d successful",
client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_type);
client->mqtt_state.pending_msg_id, client->mqtt_state.pending_msg_type);
//lock mutex
if (client->mqtt_state.pending_msg_count > 0) {
//Copy to queue buffer
@ -596,8 +606,8 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
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");
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event(client);
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event(client);
}
break;
@ -616,8 +626,8 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client)
ESP_LOGD(TAG, "received MQTT_MSG_TYPE_PUBCOMP");
if (is_valid_mqtt_msg(client, MQTT_MSG_TYPE_PUBREL, msg_id)) {
ESP_LOGD(TAG, "Receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish");
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event(client);
client->event.event_id = MQTT_EVENT_PUBLISHED;
esp_mqtt_dispatch_event(client);
}
break;
case MQTT_MSG_TYPE_PINGREQ:
@ -642,7 +652,7 @@ static void esp_mqtt_task(void *pv)
client->transport = transport_list_get_transport(client->transport_list, client->config->scheme);
if (client->transport == NULL) {
ESP_LOGE(TAG, "There are no transports valid, stop mqtt client");
ESP_LOGE(TAG, "There are no transports valid, stop mqtt client, config scheme = %s", client->config->scheme);
client->run = false;
}
//default port
@ -711,7 +721,7 @@ static void esp_mqtt_task(void *pv)
client->reconnect_tick = platform_tick_get_ms();
ESP_LOGD(TAG, "Reconnecting...");
}
vTaskDelay(client->wait_timeout_ms/2/portTICK_RATE_MS);
vTaskDelay(client->wait_timeout_ms / 2 / portTICK_RATE_MS);
break;
}
}