diff --git a/Kconfig b/Kconfig index 97e2ca2..8f3f9cd 100644 --- a/Kconfig +++ b/Kconfig @@ -9,10 +9,15 @@ config MQTT_PROTOCOL_311 config MQTT_SECURITY_ON bool "Enable MQTT over SSL" - default y + default n help Enable MQTT Over SSL +config MQTT_PRIORITY + int "MQTT Task Priority" + range 1 15 + default 5 + config CONFIG_MQTT_LOG_ERROR_ON bool "Enable MQTT Debug message" default y @@ -29,7 +34,7 @@ config CONFIG_MQTT_LOG_INFO_ON help Disable it will redurce memory and run faster config MQTT_RECONNECT_TIMEOUT - int "Reconnect timeout" + int "Reconnect timeout (in second)" range 10 16535 default 60 @@ -38,6 +43,11 @@ config MQTT_QUEUE_BUFFER_SIZE_WORD range 256 4096 default 1024 +config MQTT_BUFFER_SIZE_BYTE + int "Network buffer size for MQTT in byte" + range 128 4096 + default 1024 + config MQTT_MAX_HOST_LEN int "Maximum host name len - in byte" range 32 256 diff --git a/include/mqtt.h b/include/mqtt.h index 58b478f..a9d8d27 100644 --- a/include/mqtt.h +++ b/include/mqtt.h @@ -1,10 +1,12 @@ #ifndef _MQTT_H_ #define _MQTT_H_ #include +#include #include "mqtt_config.h" #include "mqtt_msg.h" typedef void (* mqtt_callback)(void *); + typedef struct { mqtt_callback connected_cb; mqtt_callback disconnected_cb; @@ -21,12 +23,30 @@ typedef struct { char password[CONFIG_MQTT_MAX_PASSWORD_LEN]; char lwt_topic[CONFIG_MQTT_MAX_LWT_TOPIC]; char lwt_msg[CONFIG_MQTT_MAX_LWT_MSG]; - uint32_t lwt_qos, + uint32_t lwt_qos; uint32_t lwt_retain; uint32_t clean_session; uint32_t keepalive; } mqtt_settings; +typedef struct mqtt_state_t +{ + uint16_t port; + int auto_reconnect; + mqtt_connect_info_t* connect_info; + uint8_t* in_buffer; + uint8_t* out_buffer; + int in_buffer_length; + int out_buffer_length; + uint16_t message_length; + uint16_t message_length_read; + mqtt_message_t* outbound_message; + mqtt_connection_t mqtt_connection; + uint16_t pending_msg_id; + int pending_msg_type; + int pending_publish_qos; +} mqtt_state_t; + typedef struct { int socket; mqtt_settings *settings; @@ -34,7 +54,8 @@ typedef struct { mqtt_connect_info_t connect_info; } mqtt_client; -void mqtt_task(void *pv); +void mqtt_start(mqtt_settings *mqtt_info); +void mqtt_task(void *pvParameters); void mqtt_publish(); void mqtt_subscribe(); void mqtt_detroy(); diff --git a/include/mqtt_config.h b/include/mqtt_config.h index 4f33042..7457e90 100644 --- a/include/mqtt_config.h +++ b/include/mqtt_config.h @@ -5,17 +5,17 @@ #ifdef CONFIG_MQTT_LOG_ERROR_ON - #define mqtt_error( format, ... ) printf( "[ERROR] "format, ## __VA_ARGS__ ) + #define mqtt_error( format, ... ) printf( "[MQTT ERROR] "format, ## __VA_ARGS__ ) #else #define mqtt_error( format, ... ) #endif #ifdef CONFIG_MQTT_LOG_WARN_ON - #define mqtt_warn( format, ... ) printf( "[WARN] "format, ## __VA_ARGS__ ) + #define mqtt_warn( format, ... ) printf( "[MQTT WARN] "format, ## __VA_ARGS__ ) #else #define mqtt_warn( format, ... ) #endif #ifdef CONFIG_MQTT_LOG_INFO_ON - #define mqtt_info( format, ... ) printf( "[INFO] "format, ## __VA_ARGS__ ) + #define mqtt_info( format, ... ) printf( "[MQTT INFO] "format, ## __VA_ARGS__ ) #else #define mqtt_info( format, ... ) #endif diff --git a/mqtt.c b/mqtt.c index b499084..f8744ca 100644 --- a/mqtt.c +++ b/mqtt.c @@ -2,12 +2,161 @@ * @Author: Tuan PM * @Date: 2016-09-10 09:33:06 * @Last Modified by: Tuan PM -* @Last Modified time: 2016-09-11 09:48:52 +* @Last Modified time: 2016-09-11 21:41:55 */ #include "mqtt.h" +#include "freertos/FreeRTOS.h" +#include "freertos/task.h" +#include "freertos/semphr.h" +#include "freertos/queue.h" +#include "lwip/sockets.h" +#include "lwip/dns.h" +#include "lwip/netdb.h" + +static TaskHandle_t xMqttTask = NULL; + + +static int resolev_dns(const char *host, struct sockaddr_in *ip) { + struct hostent *he; + struct in_addr **addr_list; + he = gethostbyname(host); + if (he == NULL) return 0; + addr_list = (struct in_addr **)he->h_addr_list; + if (addr_list[0] == NULL) return 0; + ip->sin_family = AF_INET; + memcpy(&ip->sin_addr, addr_list[0], sizeof(ip->sin_addr)); + return 1; +} + +static int client_connect(const char *stream_host, int stream_port) +{ + while (1) { + struct sockaddr_in remote_ip; + bzero(&remote_ip, sizeof(struct sockaddr_in)); + //if stream_host is not ip address, resolve it + if (inet_pton(AF_INET, stream_host, &(remote_ip.sin_addr)) != 1) { + if (!resolev_dns(stream_host, &remote_ip)) { + vTaskDelay(1000 / portTICK_RATE_MS); + continue; + } + } + int sock = socket(PF_INET, SOCK_STREAM, 0); + if (sock == -1) { + continue; + } + remote_ip.sin_port = htons(stream_port); + mqtt_info("Connecting to server %s...:%d,%d\n", ipaddr_ntoa((const ip_addr_t*)&remote_ip.sin_addr.s_addr), stream_port, remote_ip.sin_port); + if (connect(sock, (struct sockaddr *)(&remote_ip), sizeof(struct sockaddr)) != 00) { + close(sock); + mqtt_error("[MQTT] Conn err.\n"); + vTaskDelay(1000 / portTICK_RATE_MS); + continue; + } + return sock; + } +} +/* + * mqtt_connect + * input - client + * return 1: success, 0: fail + */ +static bool mqtt_connect(mqtt_client *client) +{ + int write_len, read_len, connect_rsp_code; + mqtt_msg_init(&client->mqtt_state.mqtt_connection, + client->mqtt_state.out_buffer, + client->mqtt_state.out_buffer_length); + client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection, + client->mqtt_state.connect_info); + client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data); + client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data, + client->mqtt_state.outbound_message->length); + mqtt_info("Sending MQTT CONNECT message, type: %d, id: %04X\r\n", + client->mqtt_state.pending_msg_type, + client->mqtt_state.pending_msg_id); + write_len = write(client->socket, + client->mqtt_state.outbound_message->data, + client->mqtt_state.outbound_message->length); + mqtt_info("Reading MQTT CONNECT response message\n"); + read_len = read(client->socket, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE); + if (read_len == 0) { + mqtt_error("Error network response\n"); + return false; + } + if (mqtt_get_type(client->mqtt_state.in_buffer) != MQTT_MSG_TYPE_CONNACK) { + mqtt_error("Invalid MSG_TYPE response: %d\n", mqtt_get_type(client->mqtt_state.in_buffer)); + return false; + } + connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer); + switch (connect_rsp_code) { + case CONNECTION_ACCEPTED: + mqtt_info("Connected\n"); + if (client->settings->connected_cb) { + client->settings->connected_cb(client); + } + return true; + case CONNECTION_REFUSE_PROTOCOL: + case CONNECTION_REFUSE_SERVER_UNAVAILABLE: + case CONNECTION_REFUSE_BAD_USERNAME: + case CONNECTION_REFUSE_NOT_AUTHORIZED: + mqtt_warn("Connection refuse, reason code: %d\r\n", connect_rsp_code); + return false; + default: + mqtt_warn("Connection refuse, Unknow reason\n"); + return false; + } + return false; +} void mqtt_task(void *pvParameters) { - mqtt_settings *settings = (mqtt_settings *)pvParameters; + mqtt_client *client = (mqtt_client *)pvParameters; + + client->socket = client_connect(client->settings->host, client->settings->port); + mqtt_info("Connected to server %s:%d\n", client->host, client->port); + + if (!mqtt_connect(client)) { + close(client->socket); + return; + } + } + +void mqtt_start(mqtt_settings *mqtt_info) +{ + if (xMqttTask != NULL) + return; + mqtt_client *client = malloc(sizeof(mqtt_client)); + if (client == NULL) { + mqtt_error("Memory not enought\n"); + return; + } + client->settings = mqtt_info; + + memset(&client->connect_info, 0, sizeof(mqtt_connect_info_t)); + client->connect_info.client_id = mqtt_info->client_id; + client->connect_info.username = mqtt_info->username; + client->connect_info.password = mqtt_info->password; + client->connect_info.keepalive = mqtt_info->keepalive; + client->connect_info.clean_session = mqtt_info->clean_session; + + client->mqtt_state.in_buffer = (uint8_t *)malloc(CONFIG_MQTT_BUFFER_SIZE_BYTE); + client->mqtt_state.in_buffer_length = CONFIG_MQTT_BUFFER_SIZE_BYTE; + client->mqtt_state.out_buffer = (uint8_t *)malloc(CONFIG_MQTT_BUFFER_SIZE_BYTE); + client->mqtt_state.out_buffer_length = CONFIG_MQTT_BUFFER_SIZE_BYTE; + client->mqtt_state.connect_info = &client->connect_info; + + mqtt_msg_init(&client->mqtt_state.mqtt_connection, + client->mqtt_state.out_buffer, + client->mqtt_state.out_buffer_length); + + + xTaskCreate(&mqtt_task, "mqtt_task", 2048, &client, CONFIG_MQTT_PRIORITY, &xMqttTask); +} + +void mqtt_stop() +{ + +} +