Merge pull request #97 from ajaybhargav/master

Multiple fixes and improvements in mqtt client source

Author: maikebing

mqttclient/mqttclient.c

@@ -65,11 +65,12 @@ typedef struct message_data {
 typedef void (*interceptor_handler_t)(void* client, message_data_t* msg);
 typedef void (*message_handler_t)(void* client, message_data_t* msg);
 typedef void (*reconnect_handler_t)(void* client, void* reconnect_date);
+typedef void (*topiclist_hanndler_t)(const char *topicname, mqtt_qos_t qos);
 
 typedef struct message_handlers {
     mqtt_list_t         list;
     mqtt_qos_t          qos;
-    const char*         topic_filter;
+    char                *topic_filter;
     message_handler_t   handler;
 } message_handlers_t;
 
@@ -111,6 +112,7 @@ typedef struct mqtt_client {
     uint32_t                    mqtt_read_buf_size;
     uint32_t                    mqtt_write_buf_size;
     uint32_t                    mqtt_reconnect_try_duration;
+    uint32_t                    mqtt_thread_stack_size;
     size_t                      mqtt_client_id_len;
     size_t                      mqtt_user_name_len;
     size_t                      mqtt_password_len;
@@ -158,6 +160,7 @@ MQTT_CLIENT_SET_STATEMENT(cmd_timeout, uint32_t)
 MQTT_CLIENT_SET_STATEMENT(read_buf_size, uint32_t)
 MQTT_CLIENT_SET_STATEMENT(write_buf_size, uint32_t)
 MQTT_CLIENT_SET_STATEMENT(reconnect_try_duration, uint32_t)
+MQTT_CLIENT_SET_STATEMENT(thread_stack_size, uint32_t)
 MQTT_CLIENT_SET_STATEMENT(reconnect_handler, reconnect_handler_t)
 MQTT_CLIENT_SET_STATEMENT(interceptor_handler, interceptor_handler_t)
 
@@ -170,7 +173,7 @@ int mqtt_keep_alive(mqtt_client_t* c);
 int mqtt_subscribe(mqtt_client_t* c, const char* topic_filter, mqtt_qos_t qos, message_handler_t msg_handler);
 int mqtt_unsubscribe(mqtt_client_t* c, const char* topic_filter);
 int mqtt_publish(mqtt_client_t* c, const char* topic_filter, mqtt_message_t* msg);
-int mqtt_list_subscribe_topic(mqtt_client_t* c);
+int mqtt_list_subscribe_topic(mqtt_client_t* c, topiclist_hanndler_t handler);
 int mqtt_set_will_options(mqtt_client_t* c, char *topic, mqtt_qos_t qos, uint8_t retained, char *message);
 
 #ifdef __cplusplus

mqttclient/mqttclient.h

@@ -1,4 +1,4 @@
-string(REGEX REPLACE ".*/(.*)" "\\1" CURRENT_LIB_NAME ${CMAKE_CURRENT_SOURCE_DIR}) 
+string(REGEX REPLACE ".*/(.*)" "\\1" CURRENT_LIB_NAME ${CMAKE_CURRENT_SOURCE_DIR})
 
 set(CURRENT_LIB_NAME ${LIBRARY_PREFIX}-${CURRENT_LIB_NAME})
 
@@ -16,7 +16,7 @@ set(${CURRENT_LIB_NAME}_INC_DIRS
 )
 
 # 源文件目录
-set(${CURRENT_LIB_NAME}_SRC_DIRS 
+set(${CURRENT_LIB_NAME}_SRC_DIRS
     ${CMAKE_CURRENT_SOURCE_DIR}
 )
 

network/CMakeLists.txt

@@ -24,7 +24,7 @@ int nettype_tcp_connect(network_t* n)
     n->socket = platform_net_socket_connect(n->host, n->port, PLATFORM_NET_PROTO_TCP);
     if (n->socket < 0)
         RETURN_ERROR(n->socket);
-    
+
     RETURN_ERROR(MQTT_SUCCESS_ERROR);
 }
 

network/nettype_tcp.c

@@ -51,9 +51,9 @@ static int nettype_tls_entropy_source(void *data, uint8_t *output, size_t len, s
 static int nettype_tls_init(network_t* n, nettype_tls_params_t* nettype_tls_params)
 {
     int rc = MQTT_SUCCESS_ERROR;
-    
+
     mbedtls_platform_set_calloc_free(platform_memory_calloc, platform_memory_free);
-    
+
     mbedtls_net_init(&(nettype_tls_params->socket_fd));
     mbedtls_ssl_init(&(nettype_tls_params->ssl));
     mbedtls_ssl_config_init(&(nettype_tls_params->ssl_conf));
@@ -99,7 +99,7 @@ static int nettype_tls_init(network_t* n, nettype_tls_params_t* nettype_tls_para
         MQTT_LOG_E("%s:%d %s()... mbedtls_ssl_conf_own_cert failed returned 0x%04x", __FILE__, __LINE__, __FUNCTION__, (rc < 0 )? -rc : rc);
         RETURN_ERROR(rc);
     }
-    
+
     mbedtls_ssl_conf_verify(&(nettype_tls_params->ssl_conf), server_certificate_verify, (void *)n->host);
 
     mbedtls_ssl_conf_authmode(&(nettype_tls_params->ssl_conf), MBEDTLS_SSL_VERIFY_REQUIRED);
@@ -130,7 +130,7 @@ int nettype_tls_connect(network_t* n)
     int rc;
     if (NULL == n)
         RETURN_ERROR(MQTT_NULL_VALUE_ERROR);
-    
+
     nettype_tls_params_t *nettype_tls_params = (nettype_tls_params_t *) platform_memory_alloc(sizeof(nettype_tls_params_t));
 
     if (NULL == nettype_tls_params)
@@ -170,12 +170,12 @@ int nettype_tls_connect(network_t* n)
 }
 
 
-void nettype_tls_disconnect(network_t* n) 
+void nettype_tls_disconnect(network_t* n)
 {
     int rc = 0;
     if (NULL == n)
         return;
-    
+
     nettype_tls_params_t *nettype_tls_params = (nettype_tls_params_t *) n->nettype_tls_params;
 
     do {
@@ -204,7 +204,7 @@ int nettype_tls_write(network_t *n, unsigned char *buf, int len, int timeout)
 
     if (NULL == n)
         RETURN_ERROR(MQTT_NULL_VALUE_ERROR);
-    
+
     nettype_tls_params_t *nettype_tls_params = (nettype_tls_params_t *) n->nettype_tls_params;
 
     platform_timer_cutdown(&timer, timeout);
@@ -217,7 +217,7 @@ int nettype_tls_write(network_t *n, unsigned char *buf, int len, int timeout)
         } else if ((rc == 0) || ((rc != MBEDTLS_ERR_SSL_WANT_WRITE) && (rc != MBEDTLS_ERR_SSL_WANT_READ) && (rc != MBEDTLS_ERR_SSL_TIMEOUT))) {
             MQTT_LOG_E("%s:%d %s()... mbedtls_ssl_write failed: 0x%04x", __FILE__, __LINE__, __FUNCTION__, (rc < 0 )? -rc : rc);
             break;
-        } 
+        }
     } while((!platform_timer_is_expired(&timer)) && (write_len < len));
 
     return write_len;
@@ -231,11 +231,11 @@ int nettype_tls_read(network_t *n, unsigned char *buf, int len, int timeout)
 
     if (NULL == n)
         RETURN_ERROR(MQTT_NULL_VALUE_ERROR);
-    
+
     nettype_tls_params_t *nettype_tls_params = (nettype_tls_params_t *) n->nettype_tls_params;
 
     platform_timer_cutdown(&timer, timeout);
-    
+
     do {
         rc = mbedtls_ssl_read(&(nettype_tls_params->ssl), (unsigned char *)(buf + read_len), len - read_len);
 
@@ -244,7 +244,7 @@ int nettype_tls_read(network_t *n, unsigned char *buf, int len, int timeout)
         } else if ((rc == 0) || ((rc != MBEDTLS_ERR_SSL_WANT_WRITE) && (rc != MBEDTLS_ERR_SSL_WANT_READ) && (rc != MBEDTLS_ERR_SSL_TIMEOUT))) {
             // MQTT_LOG_E("%s:%d %s()... mbedtls_ssl_read failed: 0x%04x", __FILE__, __LINE__, __FUNCTION__, (rc < 0 )? -rc : rc);
             break;
-        } 
+        }
     } while((!platform_timer_is_expired(&timer)) && (read_len < len));
 
     return read_len;

network/nettype_tls.c

@@ -62,3 +62,4 @@ void nettype_tls_disconnect(network_t* n);
 #endif
 
 #endif
+s

network/nettype_tls.h

@@ -91,7 +91,7 @@ int network_set_ca(network_t *n, const char *ca)
 #ifndef MQTT_NETWORK_TYPE_NO_TLS
     if ((NULL == n) || (NULL == ca))
         RETURN_ERROR(MQTT_NULL_VALUE_ERROR);
-    
+
     n->ca_crt = ca;
     n->ca_crt_len = strlen(ca);
     n->channel = NETWORK_CHANNEL_TLS;
@@ -110,4 +110,3 @@ int network_set_host_port(network_t* n, char *host, char *port)
 
     RETURN_ERROR(MQTT_SUCCESS_ERROR);
 }
-

network/network.c

@@ -1,4 +1,4 @@
-string(REGEX REPLACE ".*/(.*)" "\\1" CURRENT_LIB_NAME ${CMAKE_CURRENT_SOURCE_DIR}) 
+string(REGEX REPLACE ".*/(.*)" "\\1" CURRENT_LIB_NAME ${CMAKE_CURRENT_SOURCE_DIR})
 
 set(CURRENT_LIB_NAME ${LIBRARY_PREFIX}-${CURRENT_LIB_NAME})
 
@@ -10,7 +10,7 @@ set(${CURRENT_LIB_NAME}_INC_DIRS
 )
 
 # 源文件目录
-set(${CURRENT_LIB_NAME}_SRC_DIRS 
+set(${CURRENT_LIB_NAME}_SRC_DIRS
     ${CMAKE_CURRENT_SOURCE_DIR}/linux
 )
 
@@ -59,5 +59,3 @@ install(DIRECTORY ${${CURRENT_LIB_NAME}_INC_DIRS}
     PATTERN "*.h"
     PATTERN "*.hpp"
     PATTERN "CMakeLists.txt" EXCLUDE)
-
-

platform/CMakeLists.txt

@@ -27,6 +27,3 @@ void platform_memory_free(void *ptr)
 {
     vPortFree(ptr);
 }
-
-
-

platform/FreeRTOS/platform_memory.c

@@ -11,17 +11,17 @@ int platform_net_socket_connect(const char *host, const char *port, int proto)
 {
     int fd, ret = MQTT_SOCKET_UNKNOWN_HOST_ERROR;
     struct addrinfo hints, *addr_list, *cur;
-    
+
     /* Do name resolution with both IPv6 and IPv4 */
     memset(&hints, 0, sizeof(hints));
     hints.ai_family = AF_UNSPEC;
     hints.ai_socktype = (proto == PLATFORM_NET_PROTO_UDP) ? SOCK_DGRAM : SOCK_STREAM;
     hints.ai_protocol = (proto == PLATFORM_NET_PROTO_UDP) ? IPPROTO_UDP : IPPROTO_TCP;
-    
+
     if (getaddrinfo(host, port, &hints, &addr_list) != 0) {
         return ret;
     }
-    
+
     for (cur = addr_list; cur != NULL; cur = cur->ai_next) {
         fd = socket(cur->ai_family, cur->ai_socktype, cur->ai_protocol);
         if (fd < 0) {
@@ -51,14 +51,14 @@ int platform_net_socket_recv_timeout(int fd, unsigned char *buf, int len, int ti
 {
     int nread;
     int nleft = len;
-    unsigned char *ptr; 
+    unsigned char *ptr;
     ptr = buf;
 
     struct timeval tv = {
-        timeout / 1000, 
+        timeout / 1000,
         (timeout % 1000) * 1000
     };
-    
+
     if (tv.tv_sec < 0 || (tv.tv_sec == 0 && tv.tv_usec <= 0)) {
         tv.tv_sec = 0;
         tv.tv_usec = 100;
@@ -88,17 +88,17 @@ int platform_net_socket_write(int fd, void *buf, size_t len)
 int platform_net_socket_write_timeout(int fd, unsigned char *buf, int len, int timeout)
 {
     struct timeval tv = {
-        timeout / 1000, 
+        timeout / 1000,
         (timeout % 1000) * 1000
     };
-    
+
     if (tv.tv_sec < 0 || (tv.tv_sec == 0 && tv.tv_usec <= 0)) {
         tv.tv_sec = 0;
         tv.tv_usec = 100;
     }
 
     setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv,sizeof(struct timeval));
-    
+
     return write(fd, buf, len);
 }
 
@@ -123,4 +123,3 @@ int platform_net_socket_setsockopt(int fd, int level, int optname, const void *o
 {
     return setsockopt(fd, level, optname, optval, optlen);
 }
-