espressif/esp-mqtt
1
0
Fork 1
mirror of https://github.com/espressif/esp-mqtt.git synced 2025-07-30 10:48:06 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'client_init_refactor' into 'master'

Browse Source 
Clean client init

See merge request espressif/esp-mqtt!87
This commit is contained in:
David Čermák
2021-01-21 15:57:52 +08:00
parent 7a38867ce2 e05ed2ce18
commit a74adfa9eb
1 changed files with 142 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
mqtt_client.c
View File

@ -36,8 +36,7 @@ static const char *TAG = "MQTT_CLIENT";
ESP_EVENT_DEFINE_BASE(MQTT_EVENTS);
#endif
typedef struct mqtt_state
{
typedef struct mqtt_state {
mqtt_connect_info_t *connect_info;
uint8_t *in_buffer;
uint8_t *out_buffer;
@ -310,80 +309,60 @@ static esp_err_t esp_mqtt_check_cfg_conflict(const mqtt_config_storage_t *cfg, c
return ret;
}
static bool set_if_config(char const *const new_config, char **old_config)
{
if (new_config) {
free(*old_config);
*old_config = strdup(new_config);
if(*old_config == NULL) {
return false;
}
}
return true;
}
esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_client_config_t *config)
{
MQTT_API_LOCK(client);
//Copy user configurations to client context
esp_err_t err = ESP_OK;
mqtt_config_storage_t *cfg;
if (client->config) {
cfg = client->config;
} else {
cfg = calloc(1, sizeof(mqtt_config_storage_t));
ESP_MEM_CHECK(TAG, cfg, {
if (!client->config) {
client->config = calloc(1, sizeof(mqtt_config_storage_t));
ESP_MEM_CHECK(TAG, client->config, {
MQTT_API_UNLOCK(client);
return ESP_ERR_NO_MEM;
});
client->config = cfg;
}
if (config->task_prio) {
cfg->task_prio = config->task_prio;
}
if (cfg->task_prio <= 0) {
cfg->task_prio = MQTT_TASK_PRIORITY;
client->config->task_prio = config->task_prio;
if (client->config->task_prio <= 0) {
client->config->task_prio = MQTT_TASK_PRIORITY;
}
if (config->task_stack) {
cfg->task_stack = config->task_stack;
}
if (cfg->task_stack == 0) {
cfg->task_stack = MQTT_TASK_STACK;
client->config->task_stack = config->task_stack;
if (client->config->task_stack <= 0) {
client->config->task_stack = MQTT_TASK_STACK;
}
if (config->port) {
cfg->port = config->port;
client->config->port = config->port;
}
err = ESP_ERR_NO_MEM;
if (config->host) {
free(cfg->host);
cfg->host = strdup(config->host);
ESP_MEM_CHECK(TAG, cfg->host, goto _mqtt_set_config_failed);
}
if (config->username) {
free(client->connect_info.username);
client->connect_info.username = strdup(config->username);
ESP_MEM_CHECK(TAG, client->connect_info.username, goto _mqtt_set_config_failed);
}
if (config->password) {
free(client->connect_info.password);
client->connect_info.password = strdup(config->password);
ESP_MEM_CHECK(TAG, client->connect_info.password, goto _mqtt_set_config_failed);
}
ESP_MEM_CHECK(TAG, set_if_config(config->host, &client->config->host), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->username, &client->connect_info.username), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->password, &client->connect_info.password), goto _mqtt_set_config_failed);
if (config->client_id) {
free(client->connect_info.client_id);
client->connect_info.client_id = strdup(config->client_id);
ESP_MEM_CHECK(TAG, client->connect_info.client_id, goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->client_id, &client->connect_info.client_id), goto _mqtt_set_config_failed);
} else if (client->connect_info.client_id == NULL) {
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) {
free(cfg->uri);
cfg->uri = strdup(config->uri);
ESP_MEM_CHECK(TAG, cfg->uri, goto _mqtt_set_config_failed);
}
if (config->lwt_topic) {
free(client->connect_info.will_topic);
client->connect_info.will_topic = strdup(config->lwt_topic);
ESP_MEM_CHECK(TAG, client->connect_info.will_topic, goto _mqtt_set_config_failed);
}
ESP_MEM_CHECK(TAG, set_if_config(config->uri, &client->config->uri), goto _mqtt_set_config_failed);
ESP_MEM_CHECK(TAG, set_if_config(config->lwt_topic, &client->connect_info.will_topic), goto _mqtt_set_config_failed);
if (config->lwt_msg_len && config->lwt_msg) {
free(client->connect_info.will_message);
@ -431,100 +410,100 @@ esp_err_t esp_mqtt_set_config(esp_mqtt_client_handle_t client, const esp_mqtt_cl
#endif
}
cfg->network_timeout_ms = config->network_timeout_ms;
if (cfg->network_timeout_ms <= 0) {
cfg->network_timeout_ms = MQTT_NETWORK_TIMEOUT_MS;
client->config->network_timeout_ms = config->network_timeout_ms;
if (client->config->network_timeout_ms <= 0) {
client->config->network_timeout_ms = MQTT_NETWORK_TIMEOUT_MS;
}
if (config->user_context) {
cfg->user_context = config->user_context;
client->config->user_context = config->user_context;
}
if (config->event_handle) {
cfg->event_handle = config->event_handle;
client->config->event_handle = config->event_handle;
}
if (config->refresh_connection_after_ms) {
cfg->refresh_connection_after_ms = config->refresh_connection_after_ms;
client->config->refresh_connection_after_ms = config->refresh_connection_after_ms;
}
cfg->auto_reconnect = true;
if (config->disable_auto_reconnect == cfg->auto_reconnect) {
cfg->auto_reconnect = !config->disable_auto_reconnect;
client->config->auto_reconnect = true;
if (config->disable_auto_reconnect == client->config->auto_reconnect) {
client->config->auto_reconnect = !config->disable_auto_reconnect;
}
if (config->reconnect_timeout_ms) {
cfg->reconnect_timeout_ms = config->reconnect_timeout_ms;
client->config->reconnect_timeout_ms = config->reconnect_timeout_ms;
} else {
cfg->reconnect_timeout_ms = MQTT_RECON_DEFAULT_MS;
client->config->reconnect_timeout_ms = MQTT_RECON_DEFAULT_MS;
}
if (config->alpn_protos) {
for (int i = 0; i < cfg->num_alpn_protos; i++) {
free(cfg->alpn_protos[i]);
for (int i = 0; i < client->config->num_alpn_protos; i++) {
free(client->config->alpn_protos[i]);
}
free(cfg->alpn_protos);
cfg->num_alpn_protos = 0;
free(client->config->alpn_protos);
client->config->num_alpn_protos = 0;
const char **p;
for (p = config->alpn_protos; *p != NULL; p++ ) {
cfg->num_alpn_protos++;
client->config->num_alpn_protos++;
}
// mbedTLS expects the list to be null-terminated
cfg->alpn_protos = calloc(cfg->num_alpn_protos + 1, sizeof(config->alpn_protos));
ESP_MEM_CHECK(TAG, cfg->alpn_protos, goto _mqtt_set_config_failed);
client->config->alpn_protos = calloc(client->config->num_alpn_protos + 1, sizeof(config->alpn_protos));
ESP_MEM_CHECK(TAG, client->config->alpn_protos, goto _mqtt_set_config_failed);
for (int i = 0; i < cfg->num_alpn_protos; i++) {
cfg->alpn_protos[i] = strdup(config->alpn_protos[i]);
ESP_MEM_CHECK(TAG, cfg->alpn_protos[i], goto _mqtt_set_config_failed);
for (int i = 0; i < client->config->num_alpn_protos; i++) {
client->config->alpn_protos[i] = strdup(config->alpn_protos[i]);
ESP_MEM_CHECK(TAG, client->config->alpn_protos[i], goto _mqtt_set_config_failed);
}
}
// configure ssl related parameters
cfg->use_global_ca_store = config->use_global_ca_store;
cfg->cacert_buf = config->cert_pem;
cfg->cacert_bytes = config->cert_len;
cfg->clientcert_buf = config->client_cert_pem;
cfg->clientcert_bytes = config->client_cert_len;
cfg->clientkey_buf = config->client_key_pem;
cfg->clientkey_bytes = config->client_key_len;
cfg->psk_hint_key = config->psk_hint_key;
cfg->skip_cert_common_name_check = config->skip_cert_common_name_check;
cfg->use_secure_element = config->use_secure_element;
cfg->ds_data = config->ds_data;
client->config->use_global_ca_store = config->use_global_ca_store;
client->config->cacert_buf = config->cert_pem;
client->config->cacert_bytes = config->cert_len;
client->config->clientcert_buf = config->client_cert_pem;
client->config->clientcert_bytes = config->client_cert_len;
client->config->clientkey_buf = config->client_key_pem;
client->config->clientkey_bytes = config->client_key_len;
client->config->psk_hint_key = config->psk_hint_key;
client->config->skip_cert_common_name_check = config->skip_cert_common_name_check;
client->config->use_secure_element = config->use_secure_element;
client->config->ds_data = config->ds_data;
if (config->clientkey_password && config->clientkey_password_len) {
cfg->clientkey_password_len = config->clientkey_password_len;
cfg->clientkey_password = malloc(cfg->clientkey_password_len);
memcpy(cfg->clientkey_password, config->clientkey_password, cfg->clientkey_password_len);
client->config->clientkey_password_len = config->clientkey_password_len;
client->config->clientkey_password = malloc(client->config->clientkey_password_len);
memcpy(client->config->clientkey_password, config->clientkey_password, client->config->clientkey_password_len);
}
if (config->transport) {
free(client->config->scheme);
if (config->transport == MQTT_TRANSPORT_OVER_WS) {
cfg->scheme = create_string("ws", 2);
ESP_MEM_CHECK(TAG, cfg->scheme, goto _mqtt_set_config_failed);
client->config->scheme = create_string("ws", 2);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_set_config_failed);
} else if (config->transport == MQTT_TRANSPORT_OVER_TCP) {
cfg->scheme = create_string("mqtt", 4);
ESP_MEM_CHECK(TAG, cfg->scheme, goto _mqtt_set_config_failed);
client->config->scheme = create_string("mqtt", 4);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_set_config_failed);
} else if (config->transport == MQTT_TRANSPORT_OVER_SSL) {
cfg->scheme = create_string("mqtts", 5);
ESP_MEM_CHECK(TAG, cfg->scheme, goto _mqtt_set_config_failed);
client->config->scheme = create_string("mqtts", 5);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_set_config_failed);
} else if (config->transport == MQTT_TRANSPORT_OVER_WSS) {
cfg->scheme = create_string("wss", 3);
ESP_MEM_CHECK(TAG, cfg->scheme, goto _mqtt_set_config_failed);
client->config->scheme = create_string("wss", 3);
ESP_MEM_CHECK(TAG, client->config->scheme, goto _mqtt_set_config_failed);
}
}
// Set uri at the end of config to override separately configured uri elements
if (config->uri) {
if (esp_mqtt_client_set_uri(client, cfg->uri) != ESP_OK) {
if (esp_mqtt_client_set_uri(client, client->config->uri) != ESP_OK) {
err = ESP_FAIL;
goto _mqtt_set_config_failed;
}
}
esp_mqtt_check_cfg_conflict(cfg, config);
esp_mqtt_check_cfg_conflict(client->config, config);
MQTT_API_UNLOCK(client);
@ -660,33 +639,25 @@ static esp_err_t esp_mqtt_abort_connection(esp_mqtt_client_handle_t client)
return ESP_OK;
}
esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *config)
static bool create_client_data(esp_mqtt_client_handle_t client)
{
esp_mqtt_client_handle_t client = calloc(1, sizeof(struct esp_mqtt_client));
ESP_MEM_CHECK(TAG, client, return NULL);
client->event.error_handle = calloc(1, sizeof(esp_mqtt_error_codes_t));
if (!client->event.error_handle) {
free(client);
return NULL;
}
ESP_MEM_CHECK(TAG, client->event.error_handle, return false)
client->api_lock = xSemaphoreCreateRecursiveMutex();
if (!client->api_lock) {
free(client->event.error_handle);
free(client);
return NULL;
}
ESP_MEM_CHECK(TAG, client->api_lock, return false);
client->transport_list = esp_transport_list_init();
ESP_MEM_CHECK(TAG, client->transport_list, goto _mqtt_init_failed);
ESP_MEM_CHECK(TAG, client->transport_list, return false);
esp_transport_handle_t tcp = esp_transport_tcp_init();
ESP_MEM_CHECK(TAG, tcp, goto _mqtt_init_failed);
ESP_MEM_CHECK(TAG, tcp, return false);
esp_transport_set_default_port(tcp, MQTT_TCP_DEFAULT_PORT);
esp_transport_list_add(client->transport_list, tcp, "mqtt");
#if MQTT_ENABLE_WS
esp_transport_handle_t ws = esp_transport_ws_init(tcp);
ESP_MEM_CHECK(TAG, ws, goto _mqtt_init_failed);
ESP_MEM_CHECK(TAG, ws, return false);
esp_transport_set_default_port(ws, MQTT_WS_DEFAULT_PORT);
#ifdef MQTT_SUPPORTED_FEATURE_WS_SUBPROTOCOL
esp_transport_ws_set_subprotocol(ws, "mqtt");
@ -696,7 +667,7 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
#if MQTT_ENABLE_SSL
esp_transport_handle_t ssl = esp_transport_ssl_init();
ESP_MEM_CHECK(TAG, ssl, goto _mqtt_init_failed);
ESP_MEM_CHECK(TAG, ssl, return false);
esp_transport_set_default_port(ssl, MQTT_SSL_DEFAULT_PORT);
esp_transport_list_add(client->transport_list, ssl, "mqtts");
@ -704,14 +675,24 @@ esp_mqtt_client_handle_t esp_mqtt_client_init(const esp_mqtt_client_config_t *co
#if MQTT_ENABLE_WSS
esp_transport_handle_t wss = esp_transport_ws_init(ssl);
ESP_MEM_CHECK(TAG, wss, goto _mqtt_init_failed);
ESP_MEM_CHECK(TAG, wss, return false);
#ifdef MQTT_SUPPORTED_FEATURE_WS_SUBPROTOCOL
esp_transport_ws_set_subprotocol(wss, "mqtt");
#endif
esp_transport_set_default_port(wss, MQTT_WSS_DEFAULT_PORT);
esp_transport_list_add(client->transport_list, wss, "wss");
#endif
ESP_MEM_CHECK(TAG, client->transport_list, goto _mqtt_init_failed);
ESP_MEM_CHECK(TAG, client->transport_list, return false);
return true;
}
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));
ESP_MEM_CHECK(TAG, client, return NULL);
if (!create_client_data(client)) {
goto _mqtt_init_failed;
}
if (esp_mqtt_set_config(client, config) != ESP_OK) {
goto _mqtt_init_failed;
@ -762,7 +743,9 @@ esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client)
if (client == NULL) {
return ESP_ERR_INVALID_ARG;
}
if (client->api_lock) {
esp_mqtt_client_stop(client);
}
esp_mqtt_destroy_config(client);
if (client->transport_list) {
esp_transport_list_destroy(client->transport_list);
@ -775,7 +758,9 @@ esp_err_t esp_mqtt_client_destroy(esp_mqtt_client_handle_t client)
}
free(client->mqtt_state.in_buffer);
free(client->mqtt_state.out_buffer);
if (client->api_lock) {
vSemaphoreDelete(client->api_lock);
}
free(client->event.error_handle);
free(client);
return ESP_OK;