/* * Copyright (C) 2015-2018 Alibaba Group Holding Limited */ #include "mqtt_internal.h" #ifdef LOG_REPORT_TO_CLOUD #include "iotx_log_report.h" #endif static int _in_yield_cb; #ifndef PLATFORM_HAS_DYNMEM iotx_mc_client_t g_iotx_mc_client[IOTX_MC_CLIENT_MAX_COUNT] = {0}; #endif static void iotx_mc_release(iotx_mc_client_t *pclient) { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(pclient); #else memset(pclient, 0, sizeof(iotx_mc_client_t)); #endif } #if !WITH_MQTT_ONLY_QOS0 static void iotx_mc_pub_wait_list_init(iotx_mc_client_t *pClient) { #ifdef PLATFORM_HAS_DYNMEM INIT_LIST_HEAD(&pClient->list_pub_wait_ack); #else memset(pClient->list_pub_wait_ack, 0, sizeof(iotx_mc_pub_info_t) * IOTX_MC_PUBWAIT_LIST_MAX_LEN); #endif } static void iotx_mc_pub_wait_list_deinit(iotx_mc_client_t *pClient) { #ifdef PLATFORM_HAS_DYNMEM iotx_mc_pub_info_t *node = NULL, *next_node = NULL; linkkit_list_for_each_entry_safe(node, next_node, &pClient->list_pub_wait_ack, linked_list, iotx_mc_pub_info_t) { list_del(&node->linked_list); mqtt_free(node); } #else memset(pClient->list_pub_wait_ack, 0, sizeof(iotx_mc_pub_info_t) * IOTX_MC_PUBWAIT_LIST_MAX_LEN); #endif } #endif /* set MQTT connection parameter */ static int iotx_mc_set_connect_params(iotx_mc_client_t *pClient, MQTTPacket_connectData *pConnectParams) { if (NULL == pClient || NULL == pConnectParams) { return NULL_VALUE_ERROR; } memcpy(pClient->connect_data.struct_id, pConnectParams->struct_id, 4); pClient->connect_data.struct_version = pConnectParams->struct_version; pClient->connect_data.MQTTVersion = pConnectParams->MQTTVersion; pClient->connect_data.clientID = pConnectParams->clientID; pClient->connect_data.cleansession = pConnectParams->cleansession; pClient->connect_data.willFlag = pConnectParams->willFlag; pClient->connect_data.username = pConnectParams->username; pClient->connect_data.password = pConnectParams->password; memcpy(pClient->connect_data.will.struct_id, pConnectParams->will.struct_id, 4); pClient->connect_data.will.struct_version = pConnectParams->will.struct_version; pClient->connect_data.will.topicName = pConnectParams->will.topicName; pClient->connect_data.will.message = pConnectParams->will.message; pClient->connect_data.will.qos = pConnectParams->will.qos; pClient->connect_data.will.retained = pConnectParams->will.retained; if (pConnectParams->keepAliveInterval < CONFIG_MQTT_KEEPALIVE_INTERVAL_MIN) { mqtt_warning("Input heartbeat interval(%d ms) < Allowed minimum(%d ms)", (pConnectParams->keepAliveInterval * 1000), (CONFIG_MQTT_KEEPALIVE_INTERVAL_MIN * 1000) ); mqtt_warning("Reset heartbeat interval => %d Millisecond", (CONFIG_MQTT_KEEPALIVE_INTERVAL * 1000) ); pClient->connect_data.keepAliveInterval = CONFIG_MQTT_KEEPALIVE_INTERVAL; } else if (pConnectParams->keepAliveInterval > CONFIG_MQTT_KEEPALIVE_INTERVAL_MAX) { mqtt_warning("Input heartbeat interval(%d ms) > Allowed maximum(%d ms)", (pConnectParams->keepAliveInterval * 1000), (CONFIG_MQTT_KEEPALIVE_INTERVAL_MAX * 1000) ); mqtt_warning("Reset heartbeat interval => %d Millisecond", (CONFIG_MQTT_KEEPALIVE_INTERVAL * 1000) ); pClient->connect_data.keepAliveInterval = CONFIG_MQTT_KEEPALIVE_INTERVAL; } else { pClient->connect_data.keepAliveInterval = pConnectParams->keepAliveInterval; } return SUCCESS_RETURN; } /* set state of MQTT client */ static void iotx_mc_set_client_state(iotx_mc_client_t *pClient, iotx_mc_state_t newState) { HAL_MutexLock(pClient->lock_generic); pClient->client_state = newState; HAL_MutexUnlock(pClient->lock_generic); } static iotx_mc_state_t iotx_mc_get_client_state(iotx_mc_client_t *pClient) { iotx_mc_state_t state; HAL_MutexLock(pClient->lock_generic); state = pClient->client_state; HAL_MutexUnlock(pClient->lock_generic); return state; } /* Initialize MQTT client */ static int iotx_mc_init(iotx_mc_client_t *pClient, iotx_mqtt_param_t *pInitParams) { int rc = FAIL_RETURN; iotx_mc_state_t mc_state = IOTX_MC_STATE_INVALID; MQTTPacket_connectData connectdata = MQTTPacket_connectData_initializer; if (pClient == NULL || pInitParams == NULL || pInitParams->write_buf_size == 0 || pInitParams->read_buf_size == 0) { return NULL_VALUE_ERROR; } pClient->lock_generic = HAL_MutexCreate(); if (!pClient->lock_generic) { return FAIL_RETURN; } pClient->lock_list_pub = HAL_MutexCreate(); if (!pClient->lock_list_pub) { goto RETURN; } pClient->lock_yield = HAL_MutexCreate(); if (!pClient->lock_yield) { goto RETURN; } pClient->lock_write_buf = HAL_MutexCreate(); if (!pClient->lock_write_buf) { goto RETURN; } pClient->lock_read_buf = HAL_MutexCreate(); if (!pClient->lock_read_buf) { goto RETURN; } connectdata.MQTTVersion = IOTX_MC_MQTT_VERSION; connectdata.keepAliveInterval = pInitParams->keepalive_interval_ms / 1000; connectdata.clientID.cstring = (char *)pInitParams->client_id; connectdata.username.cstring = (char *)pInitParams->username; connectdata.password.cstring = (char *)pInitParams->password; connectdata.cleansession = pInitParams->clean_session; if (pInitParams->request_timeout_ms < CONFIG_MQTT_REQ_TIMEOUT_MIN || pInitParams->request_timeout_ms > CONFIG_MQTT_REQ_TIMEOUT_MAX) { pClient->request_timeout_ms = CONFIG_MQTT_REQUEST_TIMEOUT; } else { pClient->request_timeout_ms = pInitParams->request_timeout_ms; } #ifdef PLATFORM_HAS_DYNMEM #if !( WITH_MQTT_DYN_BUF) pClient->buf_send = mqtt_malloc(pInitParams->write_buf_size); if (pClient->buf_send == NULL) { goto RETURN; } pClient->buf_size_send = pInitParams->write_buf_size; pClient->buf_read = mqtt_malloc(pInitParams->read_buf_size); if (pClient->buf_read == NULL) { goto RETURN; } pClient->buf_size_read = pInitParams->read_buf_size; #else pClient->buf_size_send_max = pInitParams->write_buf_size; pClient->buf_size_read_max = pInitParams->read_buf_size; #endif #else pClient->buf_size_send = IOTX_MC_TX_MAX_LEN; pClient->buf_size_read = IOTX_MC_RX_MAX_LEN; #endif pClient->keepalive_probes = 0; pClient->handle_event.h_fp = pInitParams->handle_event.h_fp; pClient->handle_event.pcontext = pInitParams->handle_event.pcontext; /* Initialize reconnect parameter */ pClient->reconnect_param.reconnect_time_interval_ms = IOTX_MC_RECONNECT_INTERVAL_MIN_MS; #if !WITH_MQTT_ONLY_QOS0 iotx_mc_pub_wait_list_init(pClient); #endif #ifdef PLATFORM_HAS_DYNMEM INIT_LIST_HEAD(&pClient->list_sub_handle); INIT_LIST_HEAD(&pClient->list_sub_sync_ack); #endif /* Initialize MQTT connect parameter */ rc = iotx_mc_set_connect_params(pClient, &connectdata); if (SUCCESS_RETURN != rc) { mc_state = IOTX_MC_STATE_INVALID; goto RETURN; } iotx_time_init(&pClient->next_ping_time); iotx_time_init(&pClient->reconnect_param.reconnect_next_time); memset(&pClient->ipstack, 0, sizeof(utils_network_t)); #ifdef SUPPORT_TLS { extern const char *iotx_ca_crt; pInitParams->pub_key = iotx_ca_crt; } #endif rc = iotx_net_init(&pClient->ipstack, pInitParams->host, pInitParams->port, pInitParams->pub_key); if (SUCCESS_RETURN != rc) { mc_state = IOTX_MC_STATE_INVALID; goto RETURN; } mc_state = IOTX_MC_STATE_INITIALIZED; rc = SUCCESS_RETURN; mqtt_info("MQTT init success!"); RETURN : iotx_mc_set_client_state(pClient, mc_state); if (rc != SUCCESS_RETURN) { #ifdef PLATFORM_HAS_DYNMEM if (pClient->buf_send != NULL) { mqtt_free(pClient->buf_send); pClient->buf_send = NULL; } if (pClient->buf_read != NULL) { mqtt_free(pClient->buf_read); pClient->buf_read = NULL; } #endif if (pClient->lock_list_pub) { HAL_MutexDestroy(pClient->lock_list_pub); pClient->lock_list_pub = NULL; } if (pClient->lock_write_buf) { HAL_MutexDestroy(pClient->lock_write_buf); pClient->lock_write_buf = NULL; } if (pClient->lock_read_buf) { HAL_MutexDestroy(pClient->lock_read_buf); pClient->lock_read_buf = NULL; } if (pClient->lock_yield) { HAL_MutexDestroy(pClient->lock_yield); pClient->lock_yield = NULL; } } return rc; } #ifdef PLATFORM_HAS_DYNMEM #if WITH_MQTT_DYN_BUF extern int MQTTPacket_len(int rem_len); extern int MQTTSerialize_connectLength(MQTTPacket_connectData *options); #endif #endif static int _get_connect_length(MQTTPacket_connectData *options) { #ifdef PLATFORM_HAS_DYNMEM #if WITH_MQTT_DYN_BUF return MQTTPacket_len(MQTTSerialize_connectLength(options)); #else return 0; #endif #else return 0; #endif } static int _reset_send_buffer(iotx_mc_client_t *c) { #ifdef PLATFORM_HAS_DYNMEM #if WITH_MQTT_DYN_BUF if (c == NULL || c->buf_send == NULL) { return FAIL_RETURN; } mqtt_free(c->buf_send); c->buf_send = NULL; c->buf_size_send = 0; return 0; #else return 0; #endif #else return 0; #endif } static int _reset_recv_buffer(iotx_mc_client_t *c) { #ifdef PLATFORM_HAS_DYNMEM #if WITH_MQTT_DYN_BUF if (c == NULL || c->buf_read == NULL) { return FAIL_RETURN; } mqtt_free(c->buf_read); c->buf_read = NULL; c->buf_size_read = 0; return 0; #else return 0; #endif #else return 0; #endif } static int _alloc_send_buffer(iotx_mc_client_t *c, int len) { #ifdef PLATFORM_HAS_DYNMEM #if WITH_MQTT_DYN_BUF int tmp_len; if (c == NULL) { return FAIL_RETURN; } tmp_len = MQTT_DYNBUF_SEND_MARGIN + len; if (tmp_len > c->buf_size_send_max) { tmp_len = c->buf_size_send_max; } if (c->buf_send != NULL) { mqtt_warning("c->buf_send is not null,free it first!"); mqtt_free(c->buf_send); } c->buf_send = mqtt_malloc(tmp_len); if (c->buf_send == NULL) { return ERROR_MALLOC; } memset(c->buf_send, 0, tmp_len); c->buf_size_send = tmp_len; return SUCCESS_RETURN; #else return 0; #endif #else return 0; #endif } static int _alloc_recv_buffer(iotx_mc_client_t *c, int len) { #ifdef PLATFORM_HAS_DYNMEM #if WITH_MQTT_DYN_BUF int tmp_len; if (c == NULL) { return FAIL_RETURN; } tmp_len = MQTT_DYNBUF_RECV_MARGIN + len; if (tmp_len > c->buf_size_read_max) { tmp_len = c->buf_size_read_max; } if (c->buf_read != NULL) { /* do realloc */ char *temp = mqtt_malloc(tmp_len); if (temp == NULL) { mqtt_err("realloc err"); return ERROR_MALLOC; } memset(temp, 0, tmp_len); memcpy(temp, c->buf_read, c->buf_size_read < tmp_len ? c->buf_size_read : tmp_len); mqtt_free(c->buf_read); c->buf_read = temp; } else { c->buf_read = mqtt_malloc(tmp_len); if (c->buf_read == NULL) { mqtt_err("calloc err"); return ERROR_MALLOC; } memset(c->buf_read, 0, tmp_len); } c->buf_size_read = tmp_len; return SUCCESS_RETURN; #else return 0; #endif #else return 0; #endif } static int iotx_mc_send_packet(iotx_mc_client_t *c, char *buf, int length, iotx_time_t *time) { int rc = FAIL_RETURN; int sent = 0; unsigned int left_t = 0; if (!c || !buf || !time) { return rc; } while (sent < length && !utils_time_is_expired(time)) { left_t = iotx_time_left(time); left_t = (left_t == 0) ? 1 : left_t; rc = c->ipstack.write(&c->ipstack, &buf[sent], length - sent, left_t); if (rc < 0) { /* there was an error writing the data */ break; } sent += rc; } if (sent == length) { rc = SUCCESS_RETURN; } else { rc = MQTT_NETWORK_ERROR; } return rc; } int MQTTConnect(iotx_mc_client_t *pClient) { MQTTPacket_connectData *pConnectParams; iotx_time_t connectTimer; int len = 0; if (!pClient) { return FAIL_RETURN; } pConnectParams = &pClient->connect_data; HAL_MutexLock(pClient->lock_write_buf); len = _get_connect_length(pConnectParams); if (_alloc_send_buffer(pClient, len) != SUCCESS_RETURN) { HAL_MutexUnlock(pClient->lock_write_buf); return FAIL_RETURN; } if ((len = MQTTSerialize_connect((unsigned char *)pClient->buf_send, pClient->buf_size_send, pConnectParams)) <= 0) { mqtt_err("Serialize connect packet failed, len = %d", len); _reset_send_buffer(pClient); HAL_MutexUnlock(pClient->lock_write_buf); return MQTT_CONNECT_PACKET_ERROR; } /* send the connect packet */ iotx_time_init(&connectTimer); utils_time_countdown_ms(&connectTimer, pClient->request_timeout_ms); if ((iotx_mc_send_packet(pClient, pClient->buf_send, len, &connectTimer)) != SUCCESS_RETURN) { mqtt_err("send connect packet failed"); _reset_send_buffer(pClient); HAL_MutexUnlock(pClient->lock_write_buf); return MQTT_NETWORK_ERROR; } _reset_send_buffer(pClient); HAL_MutexUnlock(pClient->lock_write_buf); return SUCCESS_RETURN; } static int iotx_mc_decode_packet(iotx_mc_client_t *c, int *value, int timeout) { char i; int multiplier = 1; int len = 0; const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4; if (!c || !value) { return FAIL_RETURN; } *value = 0; do { int rc = MQTTPACKET_READ_ERROR; if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) { return MQTTPACKET_READ_ERROR; /* bad data */ } rc = c->ipstack.read(&c->ipstack, &i, 1, timeout == 0 ? 1 : timeout); if (rc == 0) { return FAIL_RETURN; } else if (rc != 1) { return MQTT_NETWORK_ERROR; } *value += (i & 127) * multiplier; multiplier *= 128; } while ((i & 128) != 0); return len; } static int _handle_event(iotx_mqtt_event_handle_pt handle, iotx_mc_client_t *c, iotx_mqtt_event_msg_pt msg) { if (handle == NULL || handle->h_fp == NULL) { return FAIL_RETURN; } _in_yield_cb = 1; handle->h_fp(handle->pcontext, c, msg); _in_yield_cb = 0; return 0; } static int iotx_mc_read_packet(iotx_mc_client_t *c, iotx_time_t *timer, unsigned int *packet_type) { MQTTHeader header = {0}; int len = 0; int rem_len = 0; int rc = 0; unsigned int left_t = 0; if (!c || !timer || !packet_type) { return FAIL_RETURN; } HAL_MutexLock(c->lock_read_buf); rc = _alloc_recv_buffer(c, 0); if (rc < 0) { HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } /* 1. read the header byte. This has the packet type in it */ left_t = iotx_time_left(timer); left_t = (left_t == 0) ? 1 : left_t; rc = c->ipstack.read(&c->ipstack, c->buf_read, 1, left_t); if (0 == rc) { /* timeout */ *packet_type = MQTT_CPT_RESERVED; HAL_MutexUnlock(c->lock_read_buf); return SUCCESS_RETURN; } else if (1 != rc) { mqtt_err("mqtt read error, rc=%d", rc); HAL_MutexUnlock(c->lock_read_buf); return MQTT_NETWORK_ERROR; } len = 1; /* 2. read the remaining length. This is variable in itself */ left_t = iotx_time_left(timer); left_t = (left_t == 0) ? 1 : left_t; if ((rc = iotx_mc_decode_packet(c, &rem_len, left_t)) < 0) { mqtt_err("decodePacket error,rc = %d", rc); HAL_MutexUnlock(c->lock_read_buf); return rc; } len += MQTTPacket_encode((unsigned char *)c->buf_read + 1, rem_len); /* put the original remaining length back into the buffer */ rc = _alloc_recv_buffer(c, rem_len + len); if (rc < 0) { HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } /* Check if the received data length exceeds mqtt read buffer length */ if ((rem_len > 0) && ((rem_len + len) > c->buf_size_read)) { int needReadLen; int remainDataLen; #ifdef PLATFORM_HAS_DYNMEM char *remainDataBuf; #else char remainDataBuf[IOTX_MC_RX_MAX_LEN] = {0}; #endif mqtt_err("mqtt read buffer is too short, mqttReadBufLen : %u, remainDataLen : %d", c->buf_size_read, rem_len); needReadLen = c->buf_size_read - len; left_t = iotx_time_left(timer); left_t = (left_t == 0) ? 1 : left_t; rc = c->ipstack.read(&c->ipstack, c->buf_read + len, needReadLen, left_t); if (rc < 0) { mqtt_err("mqtt read error"); HAL_MutexUnlock(c->lock_read_buf); return MQTT_NETWORK_ERROR; } else if (rc != needReadLen) { mqtt_warning("mqtt read timeout"); HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } /* drop data whitch over the length of mqtt buffer */ remainDataLen = rem_len - needReadLen; #ifdef PLATFORM_HAS_DYNMEM remainDataBuf = mqtt_malloc(remainDataLen + 1); if (!remainDataBuf) { mqtt_err("allocate remain buffer failed"); HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } #else if (remainDataLen >= IOTX_MC_RX_MAX_LEN) { mqtt_err("IOTX_MC_RX_MAX_LEN too short, remainDataLen: %d, IOTX_MC_RX_MAX_LEN: %d", remainDataLen, IOTX_MC_RX_MAX_LEN); HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } #endif left_t = iotx_time_left(timer); left_t = (left_t == 0) ? 1 : left_t; rc = c->ipstack.read(&c->ipstack, remainDataBuf, remainDataLen, left_t); if (rc < 0) { mqtt_err("mqtt read error"); #ifdef PLATFORM_HAS_DYNMEM mqtt_free(remainDataBuf); remainDataBuf = NULL; #endif HAL_MutexUnlock(c->lock_read_buf); return MQTT_NETWORK_ERROR; } else if (rc != remainDataLen) { mqtt_warning("mqtt read timeout"); #ifdef PLATFORM_HAS_DYNMEM mqtt_free(remainDataBuf); remainDataBuf = NULL; #endif HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } #ifdef PLATFORM_HAS_DYNMEM mqtt_free(remainDataBuf); remainDataBuf = NULL; #endif HAL_MutexUnlock(c->lock_read_buf); *packet_type = MQTT_CPT_RESERVED; if (NULL != c->handle_event.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_BUFFER_OVERFLOW; msg.msg = "mqtt read buffer is too short"; _handle_event(&c->handle_event, c, &msg); } return SUCCESS_RETURN; } /* 3. read the rest of the buffer using a callback to supply the rest of the data */ left_t = iotx_time_left(timer); left_t = (left_t == 0) ? 1 : left_t; rc = c->ipstack.read(&c->ipstack, c->buf_read + len, rem_len, left_t); if (rem_len > 0) { if (rc < 0) { mqtt_err("mqtt read error"); HAL_MutexUnlock(c->lock_read_buf); return MQTT_NETWORK_ERROR; } else if (rc != rem_len) { mqtt_warning("mqtt read timeout"); HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } } header.byte = c->buf_read[0]; *packet_type = MQTT_HEADER_GET_TYPE(header.byte); if ((len + rem_len) < c->buf_size_read) { c->buf_read[len + rem_len] = '\0'; } HAL_MutexUnlock(c->lock_read_buf); return SUCCESS_RETURN; } static int iotx_mc_handle_recv_CONNACK(iotx_mc_client_t *c) { int rc = SUCCESS_RETURN; unsigned char connack_rc = 255; char sessionPresent = 0; if (!c) { return FAIL_RETURN; } if (MQTTDeserialize_connack((unsigned char *)&sessionPresent, &connack_rc, (unsigned char *)c->buf_read, c->buf_size_read) != 1) { mqtt_err("connect ack is error"); return MQTT_CONNECT_ACK_PACKET_ERROR; } switch (connack_rc) { case IOTX_MC_CONNECTION_ACCEPTED: rc = SUCCESS_RETURN; break; case IOTX_MC_CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION: rc = MQTT_CONANCK_UNACCEPTABLE_PROTOCOL_VERSION_ERROR; break; case IOTX_MC_CONNECTION_REFUSED_IDENTIFIER_REJECTED: rc = MQTT_CONNACK_IDENTIFIER_REJECTED_ERROR; break; case IOTX_MC_CONNECTION_REFUSED_SERVER_UNAVAILABLE: rc = MQTT_CONNACK_SERVER_UNAVAILABLE_ERROR; break; case IOTX_MC_CONNECTION_REFUSED_BAD_USERDATA: rc = MQTT_CONNACK_BAD_USERDATA_ERROR; break; case IOTX_MC_CONNECTION_REFUSED_NOT_AUTHORIZED: rc = MQTT_CONNACK_NOT_AUTHORIZED_ERROR; break; default: rc = MQTT_CONNACK_UNKNOWN_ERROR; break; } return rc; } static int iotx_mc_wait_CONNACK(iotx_mc_client_t *c) { #define WAIT_CONNACK_MAX (10) unsigned char wait_connack = 0; unsigned int packetType = 0; int rc = 0; iotx_time_t timer; if (!c) { return FAIL_RETURN; } iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); do { /* read the socket, see what work is due */ rc = iotx_mc_read_packet(c, &timer, &packetType); if (rc != SUCCESS_RETURN) { mqtt_err("readPacket error,result = %d", rc); HAL_MutexLock(c->lock_read_buf); _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); return rc; } if (++wait_connack > WAIT_CONNACK_MAX) { mqtt_err("wait connack timeout"); HAL_MutexLock(c->lock_read_buf); _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); return MQTT_NETWORK_ERROR; } } while (packetType != CONNACK); HAL_MutexLock(c->lock_read_buf); rc = iotx_mc_handle_recv_CONNACK(c); _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); if (SUCCESS_RETURN != rc) { mqtt_err("recvConnackProc error,result = %d", rc); } return rc; } static int _mqtt_connect(void *client) { #define RETRY_TIME_LIMIT (8+1) #define RETRY_INTV_PERIOD (2000) int rc = FAIL_RETURN; int try_count = 1; iotx_mc_client_t *pClient = (iotx_mc_client_t *)client; int userKeepAliveInterval = 0; if (NULL == pClient) { return NULL_VALUE_ERROR; } userKeepAliveInterval = pClient->connect_data.keepAliveInterval; pClient->connect_data.keepAliveInterval = CONFIG_MQTT_KEEPALIVE_INTERVAL_MAX; mqtt_info("connect params: MQTTVersion=%d, clientID=%s, keepAliveInterval=%d, username=%s", pClient->connect_data.MQTTVersion, pClient->connect_data.clientID.cstring, pClient->connect_data.keepAliveInterval, pClient->connect_data.username.cstring); /* Establish TCP or TLS connection */ do { rc = MQTTConnect(pClient); pClient->connect_data.keepAliveInterval = userKeepAliveInterval; if (rc != SUCCESS_RETURN) { pClient->ipstack.disconnect(&pClient->ipstack); mqtt_err("send connect packet failed, rc = %d", rc); return rc; } rc = iotx_mc_wait_CONNACK(pClient); if (rc <= MQTT_CONNACK_NOT_AUTHORIZED_ERROR && rc >= MQTT_CONANCK_UNACCEPTABLE_PROTOCOL_VERSION_ERROR) { mqtt_err("received reject ACK from MQTT server! rc = %d", rc); pClient->ipstack.disconnect(&pClient->ipstack); return MQTT_CONNECT_ERROR; } if (SUCCESS_RETURN != rc) { mqtt_err("wait connect ACK timeout! rc = %d", rc); mqtt_warning("tried [%d/%d] times CONN, waiting for %d ms...", try_count, RETRY_TIME_LIMIT - 1, RETRY_INTV_PERIOD); HAL_SleepMs(RETRY_INTV_PERIOD); pClient->ipstack.disconnect(&pClient->ipstack); pClient->ipstack.connect(&pClient->ipstack); continue; } else { break; } } while (++try_count < RETRY_TIME_LIMIT); if (try_count == RETRY_TIME_LIMIT) { return MQTT_CONNECT_ERROR; } pClient->keepalive_probes = 0; iotx_mc_set_client_state(pClient, IOTX_MC_STATE_CONNECTED); utils_time_countdown_ms(&pClient->next_ping_time, pClient->connect_data.keepAliveInterval * 1000); mqtt_info("mqtt connect success!"); return SUCCESS_RETURN; } #if !WITH_MQTT_ONLY_QOS0 static int iotx_mc_push_pubInfo_to(iotx_mc_client_t *c, int len, unsigned short msgId, iotx_mc_pub_info_t **node) { #ifdef PLATFORM_HAS_DYNMEM int list_number; iotx_mc_pub_info_t *repubInfo; #else int idx; #endif if (!c || !node) { mqtt_err("the param of c is error!"); return FAIL_RETURN; } if ((len < 0) || (len > c->buf_size_send)) { mqtt_err("the param of len is error!"); #ifndef PLATFORM_HAS_DYNMEM if (len >= c->buf_size_send) { mqtt_err("IOTX_MC_TX_MAX_LEN is too short, len: %d, IOTX_MC_TX_MAX_LEN: %d", len, IOTX_MC_TX_MAX_LEN); } #endif return FAIL_RETURN; } #ifdef PLATFORM_HAS_DYNMEM list_number = list_entry_number(&c->list_pub_wait_ack); if (list_number >= IOTX_MC_REPUB_NUM_MAX) { mqtt_err("more than %u elements in republish list. List overflow!", list_number); return FAIL_RETURN; } repubInfo = (iotx_mc_pub_info_t *)mqtt_malloc(sizeof(iotx_mc_pub_info_t) + len); if (NULL == repubInfo) { mqtt_err("run iotx_memory_malloc is error!"); return FAIL_RETURN; } repubInfo->node_state = IOTX_MC_NODE_STATE_NORMANL; repubInfo->msg_id = msgId; repubInfo->len = len; iotx_time_start(&repubInfo->pub_start_time); repubInfo->buf = (unsigned char *)repubInfo + sizeof(iotx_mc_pub_info_t); memcpy(repubInfo->buf, c->buf_send, len); INIT_LIST_HEAD(&repubInfo->linked_list); list_add_tail(&repubInfo->linked_list, &c->list_pub_wait_ack); *node = repubInfo; return SUCCESS_RETURN; #else for (idx = 0; idx < IOTX_MC_PUBWAIT_LIST_MAX_LEN; idx++) { if (c->list_pub_wait_ack[idx].used == 0) { c->list_pub_wait_ack[idx].node_state = IOTX_MC_NODE_STATE_NORMANL; c->list_pub_wait_ack[idx].msg_id = msgId; c->list_pub_wait_ack[idx].len = len; iotx_time_start(&c->list_pub_wait_ack[idx].pub_start_time); memcpy(c->list_pub_wait_ack[idx].buf, c->buf_send, len); c->list_pub_wait_ack[idx].used = 1; *node = &c->list_pub_wait_ack[idx]; return SUCCESS_RETURN; } } mqtt_err("IOTX_MC_PUBWAIT_LIST_MAX_LEN is too short"); return FAIL_RETURN; #endif } static int iotx_mc_mask_pubInfo_from(iotx_mc_client_t *c, uint16_t msgId) { #ifdef PLATFORM_HAS_DYNMEM iotx_mc_pub_info_t *node = NULL; if (!c) { return FAIL_RETURN; } HAL_MutexLock(c->lock_list_pub); linkkit_list_for_each_entry(node, &c->list_pub_wait_ack, linked_list, iotx_mc_pub_info_t) { if (node->msg_id == msgId) { node->node_state = IOTX_MC_NODE_STATE_INVALID; /* mark as invalid node */ } } HAL_MutexUnlock(c->lock_list_pub); #else int idx; for (idx = 0; idx < IOTX_MC_PUBWAIT_LIST_MAX_LEN; idx++) { if (c->list_pub_wait_ack[idx].used && c->list_pub_wait_ack[idx].msg_id == msgId) { c->list_pub_wait_ack[idx].node_state = IOTX_MC_NODE_STATE_INVALID; /* mark as invalid node */ } } #endif return SUCCESS_RETURN; } static int MQTTRePublish(iotx_mc_client_t *c, char *buf, int len) { iotx_time_t timer; iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); HAL_MutexLock(c->lock_write_buf); if (iotx_mc_send_packet(c, buf, len, &timer) != SUCCESS_RETURN) { HAL_MutexUnlock(c->lock_write_buf); return MQTT_NETWORK_ERROR; } HAL_MutexUnlock(c->lock_write_buf); return SUCCESS_RETURN; } static int MQTTPubInfoProc(iotx_mc_client_t *pClient) { int rc = 0; iotx_mc_state_t state = IOTX_MC_STATE_INVALID; #ifdef PLATFORM_HAS_DYNMEM iotx_mc_pub_info_t *node = NULL, *next_node = NULL; #else int idx; #endif if (!pClient) { return FAIL_RETURN; } HAL_MutexLock(pClient->lock_list_pub); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry_safe(node, next_node, &pClient->list_pub_wait_ack, linked_list, iotx_mc_pub_info_t) { /* remove invalid node */ if (IOTX_MC_NODE_STATE_INVALID == node->node_state) { list_del(&node->linked_list); mqtt_free(node); continue; } state = iotx_mc_get_client_state(pClient); if (state != IOTX_MC_STATE_CONNECTED) { continue; } /* check the request if timeout or not */ if (utils_time_spend(&node->pub_start_time) <= (pClient->request_timeout_ms * 2)) { continue; } /* If wait ACK timeout, republish */ rc = MQTTRePublish(pClient, (char *)node->buf, node->len); iotx_time_start(&node->pub_start_time); if (MQTT_NETWORK_ERROR == rc) { iotx_mc_set_client_state(pClient, IOTX_MC_STATE_DISCONNECTED); break; } } #else for (idx = 0; idx < IOTX_MC_PUBWAIT_LIST_MAX_LEN; idx++) { if (pClient->list_pub_wait_ack[idx].used == 0) { continue; } if (IOTX_MC_NODE_STATE_INVALID == pClient->list_pub_wait_ack[idx].node_state) { memset(&pClient->list_pub_wait_ack[idx], 0, sizeof(iotx_mc_pub_info_t)); continue; } state = iotx_mc_get_client_state(pClient); if (state != IOTX_MC_STATE_CONNECTED) { continue; } /* check the request if timeout or not */ if (utils_time_spend(&pClient->list_pub_wait_ack[idx].pub_start_time) <= (pClient->request_timeout_ms * 2)) { continue; } /* If wait ACK timeout, republish */ rc = MQTTRePublish(pClient, (char *)pClient->list_pub_wait_ack[idx].buf, pClient->list_pub_wait_ack[idx].len); iotx_time_start(&pClient->list_pub_wait_ack[idx].pub_start_time); if (MQTT_NETWORK_ERROR == rc) { iotx_mc_set_client_state(pClient, IOTX_MC_STATE_DISCONNECTED); break; } } #endif HAL_MutexUnlock(pClient->lock_list_pub); return SUCCESS_RETURN; } /* handle PUBACK packet received from remote MQTT broker */ static int iotx_mc_handle_recv_PUBACK(iotx_mc_client_t *c) { unsigned short mypacketid; unsigned char dup = 0; unsigned char type = 0; if (!c) { return FAIL_RETURN; } if (MQTTDeserialize_ack(&type, &dup, &mypacketid, (unsigned char *)c->buf_read, c->buf_size_read) != 1) { return MQTT_PUBLISH_ACK_PACKET_ERROR; } (void)iotx_mc_mask_pubInfo_from(c, mypacketid); /* call callback function to notify that PUBLISH is successful */ if (NULL != c->handle_event.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_PUBLISH_SUCCESS; msg.msg = (void *)(uintptr_t)mypacketid; _handle_event(&c->handle_event, c, &msg); } return SUCCESS_RETURN; } #endif static void _iotx_mqtt_event_handle_sub(void *pcontext, void *pclient, iotx_mqtt_event_msg_pt msg) { iotx_mc_client_t *client; uintptr_t packet_id; #ifdef PLATFORM_HAS_DYNMEM mqtt_sub_sync_node_t *node = NULL; mqtt_sub_sync_node_t *next = NULL; #else int idx; #endif if (pclient == NULL || msg == NULL) { return; } client = (iotx_mc_client_t *)pclient; packet_id = (uintptr_t) msg->msg; mqtt_debug("packet_id = %lu, event_type=%d", packet_id, msg->event_type); HAL_MutexLock(client->lock_generic); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry_safe(node, next, &client->list_sub_sync_ack, linked_list, mqtt_sub_sync_node_t) { if (node->packet_id == packet_id) { node->ack_type = msg->event_type; } } #else for (idx = 0; idx < IOTX_MC_SUBSYNC_LIST_MAX_LEN; idx++) { if (client->list_sub_sync_ack[idx].used && client->list_sub_sync_ack[idx].packet_id == packet_id) { client->list_sub_sync_ack[idx].ack_type = msg->event_type; } } #endif HAL_MutexUnlock(client->lock_generic); } static int iotx_mc_handle_recv_SUBACK(iotx_mc_client_t *c) { unsigned short mypacketid; iotx_mqtt_event_msg_t msg; int i = 0, count = 0, fail_flag = -1, j = 0; int grantedQoS[MUTLI_SUBSCIRBE_MAX]; int rc; if (!c) { return FAIL_RETURN; } rc = MQTTDeserialize_suback(&mypacketid, MUTLI_SUBSCIRBE_MAX, &count, grantedQoS, (unsigned char *)c->buf_read, c->buf_size_read); if (rc < 0) { mqtt_err("Sub ack packet error, rc = MQTTDeserialize_suback() = %d", rc); return MQTT_SUBSCRIBE_ACK_PACKET_ERROR; } mqtt_debug("%20s : %d", "Return Value", rc); mqtt_debug("%20s : %d", "Packet ID", mypacketid); mqtt_debug("%20s : %d", "Count", count); for (i = 0; i < count; ++i) { mqtt_debug("%16s[%02d] : %d", "Granted QoS", i, grantedQoS[i]); } for (j = 0; j < count; j++) { fail_flag = 0; /* In negative case, grantedQoS will be 0xFFFF FF80, which means -128 */ if ((uint8_t)grantedQoS[j] == 0x80) { fail_flag = 1; mqtt_err("MQTT SUBSCRIBE failed, ack code is 0x80"); } } /* call callback function to notify that SUBSCRIBE is successful */ msg.msg = (void *)(uintptr_t)mypacketid; if (fail_flag == 1) { msg.event_type = IOTX_MQTT_EVENT_SUBCRIBE_NACK; } else { msg.event_type = IOTX_MQTT_EVENT_SUBCRIBE_SUCCESS; } _iotx_mqtt_event_handle_sub(c->handle_event.pcontext, c, &msg); if (NULL != c->handle_event.h_fp) { _handle_event(&c->handle_event, c, &msg); } return SUCCESS_RETURN; } #if WITH_MQTT_ZIP_TOPIC #define MQTT_ZIP_PATH_DEFAULT_LEN (32) static int iotx_mc_get_zip_topic(const char *path, int len, char outbuf[], int outlen) { unsigned char comp_data[MQTT_ZIP_PATH_DEFAULT_LEN] = {0}; if (!path || !len || !outbuf || !outlen) { return -1; } utils_sha256((unsigned char *)path, (size_t)len, comp_data); memcpy(outbuf, comp_data, outlen > MQTT_ZIP_PATH_DEFAULT_LEN ? MQTT_ZIP_PATH_DEFAULT_LEN : outlen); return 0; } #endif static char iotx_mc_is_topic_matched(char *topicFilter, MQTTString *topicName) { char *curf; char *curn; char *curn_end; if (!topicFilter || !topicName) { return 0; } curf = topicFilter; curn = topicName->lenstring.data; curn_end = curn + topicName->lenstring.len; while (*curf && curn < curn_end) { if (*curn == '/' && *curf != '/') { break; } if (*curf != '+' && *curf != '#' && *curf != *curn) { break; } if (*curf == '+') { /* skip until we meet the next separator, or end of string */ char *nextpos = curn + 1; while (nextpos < curn_end && *nextpos != '/') { nextpos = ++curn + 1; } } else if (*curf == '#') { curn = curn_end - 1; /* skip until end of string */ } curf++; curn++; } return (curn == curn_end) && (*curf == '\0'); } static void iotx_mc_deliver_message(iotx_mc_client_t *c, MQTTString *topicName, iotx_mqtt_topic_info_pt topic_msg) { int flag_matched = 0; MQTTString *compare_topic = NULL; #ifdef PLATFORM_HAS_DYNMEM iotx_mc_topic_handle_t *node = NULL; #else int idx = 0; #endif #if WITH_MQTT_ZIP_TOPIC MQTTString md5_topic; char md5_topic_data[MQTT_ZIP_PATH_DEFAULT_LEN] = {0}; char *net_topic; uint32_t net_topic_len; #endif if (!c || !topicName || !topic_msg) { return; } topic_msg->ptopic = topicName->lenstring.data; topic_msg->topic_len = topicName->lenstring.len; #if WITH_MQTT_ZIP_TOPIC if (topicName->cstring) { net_topic = topicName->cstring; net_topic_len = strlen(topicName->cstring); } else { net_topic = topicName->lenstring.data; net_topic_len = topicName->lenstring.len; } md5_topic.cstring = NULL; md5_topic.lenstring.data = md5_topic_data; md5_topic.lenstring.len = MQTT_ZIP_PATH_DEFAULT_LEN; iotx_mc_get_zip_topic(net_topic, net_topic_len, md5_topic_data, MQTT_ZIP_PATH_DEFAULT_LEN); compare_topic = &md5_topic; #else compare_topic = topicName; #endif /* we have to find the right message handler - indexed by topic */ HAL_MutexLock(c->lock_generic); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry(node, &c->list_sub_handle, linked_list, iotx_mc_topic_handle_t) { if (MQTTPacket_equals(compare_topic, (char *)node->topic_filter) || iotx_mc_is_topic_matched((char *)node->topic_filter, topicName)) { mqtt_debug("topic be matched"); HAL_MutexUnlock(c->lock_generic); if (NULL != node->handle.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_PUBLISH_RECEIVED; msg.msg = (void *)topic_msg; _handle_event(&node->handle, c, &msg); flag_matched = 1; } HAL_MutexLock(c->lock_generic); } } #else for (idx = 0; idx < IOTX_MC_SUBHANDLE_LIST_MAX_LEN; idx++) { if ((c->list_sub_handle[idx].used == 1) && (MQTTPacket_equals(compare_topic, (char *)c->list_sub_handle[idx].topic_filter) || iotx_mc_is_topic_matched((char *)c->list_sub_handle[idx].topic_filter, topicName))) { mqtt_debug("topic be matched"); HAL_MutexUnlock(c->lock_generic); if (NULL != c->list_sub_handle[idx].handle.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_PUBLISH_RECEIVED; msg.msg = (void *)topic_msg; _handle_event(&c->list_sub_handle[idx].handle, c, &msg); flag_matched = 1; } HAL_MutexLock(c->lock_generic); } } #endif HAL_MutexUnlock(c->lock_generic); if (0 == flag_matched) { mqtt_info("NO matching any topic, call default handle function"); if (NULL != c->handle_event.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_PUBLISH_RECEIVED; msg.msg = topic_msg; _handle_event(&c->handle_event, c, &msg); } } } static int MQTTPuback(iotx_mc_client_t *c, unsigned int msgId, enum msgTypes type) { int rc = 0; int len = 0; iotx_time_t timer; if (!c) { return FAIL_RETURN; } iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); HAL_MutexLock(c->lock_write_buf); if (type == PUBACK) { if (_alloc_send_buffer(c, 0) < 0) { HAL_MutexUnlock(c->lock_write_buf); return FAIL_RETURN; } len = MQTTSerialize_ack((unsigned char *)c->buf_send, c->buf_size_send, PUBACK, 0, msgId); #if WITH_MQTT_QOS2_PACKET } else if (type == PUBREC) { if (_alloc_send_buffer(c, 0) < 0) { HAL_MutexUnlock(c->lock_write_buf); return FAIL_RETURN; } len = MQTTSerialize_ack((unsigned char *)c->buf_send, c->buf_size_send, PUBREC, 0, msgId); } else if (type == PUBREL) { if (_alloc_send_buffer(c, 0) < 0) { HAL_MutexUnlock(c->lock_write_buf); return FAIL_RETURN; } len = MQTTSerialize_ack((unsigned char *)c->buf_send, c->buf_size_send, PUBREL, 0, msgId); #endif /* #if WITH_MQTT_QOS2_PACKET */ } else { HAL_MutexUnlock(c->lock_write_buf); return MQTT_PUBLISH_ACK_TYPE_ERROR; } if (len <= 0) { _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return MQTT_PUBLISH_ACK_PACKET_ERROR; } rc = iotx_mc_send_packet(c, c->buf_send, len, &timer); if (rc != SUCCESS_RETURN) { _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return MQTT_NETWORK_ERROR; } _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return SUCCESS_RETURN; } static int iotx_mc_handle_recv_PUBLISH(iotx_mc_client_t *c) { int result = 0; MQTTString topicName; iotx_mqtt_topic_info_t topic_msg; int qos = 0; uint32_t payload_len = 0; #ifdef INFRA_LOG_NETWORK_PAYLOAD const char *json_payload = NULL; #endif if (!c) { return FAIL_RETURN; } memset(&topic_msg, 0x0, sizeof(iotx_mqtt_topic_info_t)); memset(&topicName, 0x0, sizeof(MQTTString)); if (1 != MQTTDeserialize_publish((unsigned char *)&topic_msg.dup, (int *)&qos, (unsigned char *)&topic_msg.retain, (unsigned short *)&topic_msg.packet_id, &topicName, (unsigned char **)&topic_msg.payload, (int *)&payload_len, (unsigned char *)c->buf_read, c->buf_size_read)) { return MQTT_PUBLISH_PACKET_ERROR; } topic_msg.qos = (unsigned char)qos; topic_msg.payload_len = payload_len; if (topicName.lenstring.len == 0 || topicName.lenstring.data == NULL) { mqtt_err("Null topicName"); return MQTT_PUBLISH_PACKET_ERROR; } #ifdef INFRA_LOG_NETWORK_PAYLOAD json_payload = (const char *)topic_msg.payload; mqtt_info("Downstream Topic: '%.*s'", topicName.lenstring.len, topicName.lenstring.data); mqtt_info("Downstream Payload:"); iotx_facility_json_print(json_payload, LOG_INFO_LEVEL, '<'); #endif /* #ifdef INFRA_LOG */ mqtt_debug("%20s : %08d", "Packet Ident", topic_msg.packet_id); mqtt_debug("%20s : %d", "Topic Length", topicName.lenstring.len); mqtt_debug("%20s : %.*s", "Topic Name", topicName.lenstring.len, topicName.lenstring.data); mqtt_debug("%20s : %u / %d", "Payload Len/Room", (unsigned int)topic_msg.payload_len, (int)(c->buf_read + c->buf_size_read - topic_msg.payload)); mqtt_debug("%20s : %lu", "Receive Buflen", c->buf_size_read); #if defined(INSPECT_MQTT_FLOW) mqtt_debug("%20s : %p", "Payload Buffer", topic_msg.payload); mqtt_debug("%20s : %p", "Receive Buffer", c->buf_read); #ifdef INFRA_LOG HEXDUMP_DEBUG(topic_msg.payload, topic_msg.payload_len); #endif #endif #ifdef LOG_REPORT_TO_CLOUD get_msgid(topicName.lenstring.data, 1); #endif topic_msg.ptopic = NULL; topic_msg.topic_len = 0; mqtt_debug("delivering msg ..."); #if WITH_MQTT_FLOW_CTRL /* flowControl for specific topic */ static uint64_t time_prev = 0; uint64_t time_curr = 0; char *filterStr = "{\"method\":\"thing.service.property.set\""; int filterLen = strlen(filterStr); if (0 == memcmp(topic_msg.payload, filterStr, filterLen)) { time_curr = HAL_UptimeMs(); if (time_curr < time_prev) { time_curr = time_prev; } if ((time_curr - time_prev) <= (uint64_t)50) { mqtt_info("MQTT over threshould"); return SUCCESS_RETURN; } else { time_prev = time_curr; } } #endif iotx_mc_deliver_message(c, &topicName, &topic_msg); if (topic_msg.qos == IOTX_MQTT_QOS0) { return SUCCESS_RETURN; } else if (topic_msg.qos == IOTX_MQTT_QOS1) { result = MQTTPuback(c, topic_msg.packet_id, PUBACK); } else if (topic_msg.qos == IOTX_MQTT_QOS2) { result = MQTTPuback(c, topic_msg.packet_id, PUBREC); } else { mqtt_err("Invalid QOS, QOSvalue = %d", topic_msg.qos); return MQTT_PUBLISH_QOS_ERROR; } return result; } static int iotx_mc_handle_recv_UNSUBACK(iotx_mc_client_t *c) { unsigned short mypacketid = 0; /* should be the same as the packetid above */ if (!c) { return FAIL_RETURN; } if (MQTTDeserialize_unsuback(&mypacketid, (unsigned char *)c->buf_read, c->buf_size_read) != 1) { return MQTT_UNSUBSCRIBE_ACK_PACKET_ERROR; } if (NULL != c->handle_event.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_UNSUBCRIBE_SUCCESS; msg.msg = (void *)(uintptr_t)mypacketid; _handle_event(&c->handle_event, c, &msg); } return SUCCESS_RETURN; } static int iotx_mc_cycle(iotx_mc_client_t *c, iotx_time_t *timer) { unsigned int packetType; iotx_mc_state_t state; int rc = SUCCESS_RETURN; if (!c) { return FAIL_RETURN; } state = iotx_mc_get_client_state(c); if (state != IOTX_MC_STATE_CONNECTED) { mqtt_debug("state = %d", state); return MQTT_STATE_ERROR; } if (IOTX_MC_KEEPALIVE_PROBE_MAX < c->keepalive_probes) { iotx_mc_set_client_state(c, IOTX_MC_STATE_DISCONNECTED); c->keepalive_probes = 0; mqtt_debug("keepalive_probes more than %u, disconnected\n", IOTX_MC_KEEPALIVE_PROBE_MAX); } /* read the socket, see what work is due */ rc = iotx_mc_read_packet(c, timer, &packetType); if (rc != SUCCESS_RETURN) { HAL_MutexLock(c->lock_read_buf); _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); if (rc == MQTT_NETWORK_ERROR) { iotx_mc_set_client_state(c, IOTX_MC_STATE_DISCONNECTED); } mqtt_err("readPacket error,result = %d", rc); return MQTT_NETWORK_ERROR; } if (MQTT_CPT_RESERVED == packetType) { /* mqtt_debug("wait data timeout"); */ HAL_MutexLock(c->lock_read_buf); _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); return SUCCESS_RETURN; } /* clear ping mark when any data received from MQTT broker */ HAL_MutexLock(c->lock_generic); c->keepalive_probes = 0; HAL_MutexUnlock(c->lock_generic); HAL_MutexLock(c->lock_read_buf); switch (packetType) { case CONNACK: { mqtt_debug("CONNACK"); break; } #if !WITH_MQTT_ONLY_QOS0 case PUBACK: { mqtt_debug("PUBACK"); rc = iotx_mc_handle_recv_PUBACK(c); if (SUCCESS_RETURN != rc) { mqtt_err("recvPubackProc error,result = %d", rc); } break; } #endif case SUBACK: { mqtt_debug("SUBACK"); rc = iotx_mc_handle_recv_SUBACK(c); if (SUCCESS_RETURN != rc) { mqtt_err("recvSubAckProc error,result = %d", rc); } break; } case PUBLISH: { mqtt_debug("PUBLISH"); /* HEXDUMP_DEBUG(c->buf_read, 32); */ rc = iotx_mc_handle_recv_PUBLISH(c); if (SUCCESS_RETURN != rc) { mqtt_err("recvPublishProc error,result = %d", rc); } break; } case UNSUBACK: { mqtt_debug("UNSUBACK"); rc = iotx_mc_handle_recv_UNSUBACK(c); if (SUCCESS_RETURN != rc) { mqtt_err("recvUnsubAckProc error,result = %d", rc); } break; } case PINGRESP: { rc = SUCCESS_RETURN; mqtt_info("receive ping response!"); break; } default: mqtt_err("INVALID TYPE"); _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); return FAIL_RETURN; } _reset_recv_buffer(c); HAL_MutexUnlock(c->lock_read_buf); return rc; } void _mqtt_cycle(void *client) { int rc = SUCCESS_RETURN; iotx_time_t time; iotx_mc_client_t *pClient = (iotx_mc_client_t *)client; iotx_time_init(&time); utils_time_countdown_ms(&time, pClient->cycle_timeout_ms); do { unsigned int left_t; if (SUCCESS_RETURN != rc) { mqtt_err("error occur rc=%d", rc); } HAL_MutexLock(pClient->lock_yield); /* acquire package in cycle, such as PINGRESP or PUBLISH */ rc = iotx_mc_cycle(pClient, &time); if (SUCCESS_RETURN == rc) { #ifndef ASYNC_PROTOCOL_STACK #if !WITH_MQTT_ONLY_QOS0 /* check list of wait publish ACK to remove node that is ACKED or timeout */ MQTTPubInfoProc(pClient); #endif #endif } HAL_MutexUnlock(pClient->lock_yield); left_t = iotx_time_left(&time); if (left_t < 10) { HAL_SleepMs(left_t); } else { HAL_SleepMs(10); } } while (!utils_time_is_expired(&time)); } static int MQTTKeepalive(iotx_mc_client_t *pClient) { int len = 0; int rc = 0; /* there is no ping outstanding - send ping packet */ iotx_time_t timer; if (!pClient) { return FAIL_RETURN; } iotx_time_init(&timer); utils_time_countdown_ms(&timer, 1000); HAL_MutexLock(pClient->lock_write_buf); rc = _alloc_send_buffer(pClient, 0); if (rc < 0) { HAL_MutexUnlock(pClient->lock_write_buf); return FAIL_RETURN; } len = MQTTSerialize_pingreq((unsigned char *)pClient->buf_send, pClient->buf_size_send); mqtt_debug("len = MQTTSerialize_pingreq() = %d", len); if (len <= 0) { mqtt_err("Serialize ping request is error"); _reset_send_buffer(pClient); HAL_MutexUnlock(pClient->lock_write_buf); return MQTT_PING_PACKET_ERROR; } rc = iotx_mc_send_packet(pClient, pClient->buf_send, len, &timer); if (SUCCESS_RETURN != rc) { /* ping outstanding, then close socket unsubscribe topic and handle callback function */ mqtt_err("ping outstanding is error,result = %d", rc); _reset_send_buffer(pClient); HAL_MutexUnlock(pClient->lock_write_buf); return MQTT_NETWORK_ERROR; } _reset_send_buffer(pClient); HAL_MutexUnlock(pClient->lock_write_buf); return SUCCESS_RETURN; } static int iotx_mc_keepalive_sub(iotx_mc_client_t *pClient) { int rc = SUCCESS_RETURN; if (NULL == pClient) { return NULL_VALUE_ERROR; } /* if in disabled state, without having to send ping packets */ if (!wrapper_mqtt_check_state(pClient)) { return SUCCESS_RETURN; } /* if there is no ping_timer timeout, then return success */ if (!utils_time_is_expired(&pClient->next_ping_time)) { return SUCCESS_RETURN; } /* update to next time sending MQTT keep-alive */ utils_time_countdown_ms(&pClient->next_ping_time, pClient->connect_data.keepAliveInterval * 1000); rc = MQTTKeepalive(pClient); if (SUCCESS_RETURN != rc) { if (rc == MQTT_NETWORK_ERROR) { iotx_mc_set_client_state(pClient, IOTX_MC_STATE_DISCONNECTED); } mqtt_err("ping outstanding is error,result = %d", rc); return rc; } mqtt_info("send MQTT ping..."); HAL_MutexLock(pClient->lock_generic); pClient->keepalive_probes++; HAL_MutexUnlock(pClient->lock_generic); return SUCCESS_RETURN; } static int iotx_mc_attempt_reconnect(iotx_mc_client_t *pClient) { int rc; if (pClient == NULL) { return NULL_VALUE_ERROR; } pClient->ipstack.disconnect(&pClient->ipstack); mqtt_info("reconnect params: MQTTVersion=%d, clientID=%s, keepAliveInterval=%d, username=%s", pClient->connect_data.MQTTVersion, pClient->connect_data.clientID.cstring, pClient->connect_data.keepAliveInterval, pClient->connect_data.username.cstring); /* Ignoring return code. failures expected if network is disconnected */ rc = wrapper_mqtt_connect(pClient); if (SUCCESS_RETURN != rc && MQTT_CONNECT_BLOCK != rc) { mqtt_err("run iotx_mqtt_connect() error!"); } return rc; } static int iotx_mc_handle_reconnect(iotx_mc_client_t *pClient) { int rc = FAIL_RETURN; uint32_t interval_ms = 0; if (NULL == pClient) { return NULL_VALUE_ERROR; } mqtt_info("Waiting to reconnect..."); if (!utils_time_is_expired(&(pClient->reconnect_param.reconnect_next_time))) { /* Timer has not expired. Not time to attempt reconnect yet. Return attempting reconnect */ HAL_SleepMs(100); return FAIL_RETURN; } mqtt_info("start to reconnect"); /* rc = _conn_info_dynamic_reload(pClient); if (SUCCESS_RETURN != rc) { mqtt_err("update connect info err"); return -1; } */ rc = iotx_mc_attempt_reconnect(pClient); if (SUCCESS_RETURN == rc) { iotx_mc_set_client_state(pClient, IOTX_MC_STATE_CONNECTED); /* _conn_info_dynamic_reload_clear(pClient); */ return SUCCESS_RETURN; } else if (MQTT_CONNECT_BLOCK == rc) { return rc; } else { /* if reconnect network failed, then increase currentReconnectWaitInterval */ /* e.g. init currentReconnectWaitInterval=1s, reconnect failed, then 2s..4s..8s */ if (IOTX_MC_RECONNECT_INTERVAL_MAX_MS > pClient->reconnect_param.reconnect_time_interval_ms) { pClient->reconnect_param.reconnect_time_interval_ms *= 2; } else { pClient->reconnect_param.reconnect_time_interval_ms = IOTX_MC_RECONNECT_INTERVAL_MAX_MS; } } /* _conn_info_dynamic_reload_clear(pClient); */ interval_ms = pClient->reconnect_param.reconnect_time_interval_ms; if (IOTX_MC_RECONNECT_INTERVAL_MAX_MS < interval_ms) { interval_ms = IOTX_MC_RECONNECT_INTERVAL_MAX_MS; } utils_time_countdown_ms(&(pClient->reconnect_param.reconnect_next_time), interval_ms); mqtt_err("mqtt reconnect failed rc = %d", rc); return rc; } static void iotx_mc_reconnect_callback(iotx_mc_client_t *pClient) { /* handle callback function */ if (NULL != pClient->handle_event.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_RECONNECT; msg.msg = NULL; pClient->handle_event.h_fp(pClient->handle_event.pcontext, pClient, &msg); } } static void iotx_mc_disconnect_callback(iotx_mc_client_t *pClient) { if (NULL != pClient->handle_event.h_fp) { iotx_mqtt_event_msg_t msg; msg.event_type = IOTX_MQTT_EVENT_DISCONNECT; msg.msg = NULL; pClient->handle_event.h_fp(pClient->handle_event.pcontext, pClient, &msg); } } static void iotx_mc_keepalive(iotx_mc_client_t *pClient) { int rc = 0; iotx_mc_state_t currentState; if (!pClient) { return; } /* Periodic sending ping packet to detect whether the network is connected */ iotx_mc_keepalive_sub(pClient); currentState = iotx_mc_get_client_state(pClient); do { /* if Exceeds the maximum delay time, then return reconnect timeout */ if (IOTX_MC_STATE_DISCONNECTED_RECONNECTING == currentState || IOTX_MC_STATE_CONNECT_BLOCK == currentState) { /* Reconnection is successful, Resume regularly ping packets */ rc = iotx_mc_handle_reconnect(pClient); if (SUCCESS_RETURN != rc) { mqtt_err("reconnect network fail, rc = %d", rc); } else if (MQTT_CONNECT_BLOCK == rc) { mqtt_debug("now using async protocol stack, wait network connected..."); } else { mqtt_info("network is reconnected!"); iotx_mc_reconnect_callback(pClient); pClient->reconnect_param.reconnect_time_interval_ms = IOTX_MC_RECONNECT_INTERVAL_MIN_MS; } break; } /* If network suddenly interrupted, stop pinging packet, try to reconnect network immediately */ if (IOTX_MC_STATE_DISCONNECTED == currentState) { mqtt_err("network is disconnected!"); iotx_mc_disconnect_callback(pClient); pClient->reconnect_param.reconnect_time_interval_ms = IOTX_MC_RECONNECT_INTERVAL_MIN_MS; utils_time_countdown_ms(&(pClient->reconnect_param.reconnect_next_time), pClient->reconnect_param.reconnect_time_interval_ms); pClient->ipstack.disconnect(&pClient->ipstack); iotx_mc_set_client_state(pClient, IOTX_MC_STATE_DISCONNECTED_RECONNECTING); break; } } while (0); } static int iotx_mc_check_handle_is_identical_ex(iotx_mc_topic_handle_t *messageHandlers1, iotx_mc_topic_handle_t *messageHandler2) { int topicNameLen = 0; if (!messageHandlers1 || !messageHandler2) { return 1; } if (!(messageHandlers1->topic_filter) || !(messageHandler2->topic_filter)) { return 1; } #if !(WITH_MQTT_ZIP_TOPIC) topicNameLen = strlen(messageHandlers1->topic_filter); if (topicNameLen != strlen(messageHandler2->topic_filter)) { return 1; } if (0 != strncmp(messageHandlers1->topic_filter, messageHandler2->topic_filter, topicNameLen)) { return 1; } #else if (messageHandlers1->topic_type != messageHandler2->topic_type) { return 1; } if (messageHandlers1->topic_type == TOPIC_NAME_TYPE) { int i; for (i = 0; i < MQTT_ZIP_PATH_DEFAULT_LEN; i++) { if (messageHandler2->topic_filter[i] != messageHandlers1->topic_filter[i]) { return 1; } } } else { topicNameLen = strlen(messageHandlers1->topic_filter); if (topicNameLen != strlen(messageHandler2->topic_filter)) { return 1; } if (0 != strncmp(messageHandlers1->topic_filter, messageHandler2->topic_filter, topicNameLen)) { return 1; } } #endif return 0; } static int iotx_mc_check_handle_is_identical(iotx_mc_topic_handle_t *messageHandlers1, iotx_mc_topic_handle_t *messageHandler2) { if (iotx_mc_check_handle_is_identical_ex(messageHandlers1, messageHandler2) != 0) { return 1; } if (messageHandlers1->handle.h_fp != messageHandler2->handle.h_fp) { return 1; } /* context must be identical also */ if (messageHandlers1->handle.pcontext != messageHandler2->handle.pcontext) { return 1; } return 0; } static int MQTTSubscribe(iotx_mc_client_t *c, const char *topicFilter, iotx_mqtt_qos_t qos, unsigned int msgId, iotx_mqtt_event_handle_func_fpt messageHandler, void *pcontext) { int len = 0; iotx_time_t timer; MQTTString topic = MQTTString_initializer; /*iotx_mc_topic_handle_t handler = {topicFilter, {messageHandler, pcontext}};*/ iotx_mc_topic_handle_t *handler = NULL; #ifndef PLATFORM_HAS_DYNMEM int idx = 0; #endif if (!c || !topicFilter || !messageHandler) { return FAIL_RETURN; } #if !( WITH_MQTT_DYN_BUF) if (!c->buf_send) { return FAIL_RETURN; } #endif topic.cstring = (char *)topicFilter; iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); #ifdef PLATFORM_HAS_DYNMEM handler = mqtt_malloc(sizeof(iotx_mc_topic_handle_t)); if (NULL == handler) { return FAIL_RETURN; } memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); INIT_LIST_HEAD(&handler->linked_list); #else for (idx = 0; idx < IOTX_MC_SUBHANDLE_LIST_MAX_LEN; idx++) { if (c->list_sub_handle[idx].used == 0) { handler = &c->list_sub_handle[idx]; memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); c->list_sub_handle[idx].used = 1; break; } } if (handler == NULL) { return MQTT_SUBHANDLE_LIST_LEN_TOO_SHORT; } #endif #if !(WITH_MQTT_ZIP_TOPIC) #ifdef PLATFORM_HAS_DYNMEM handler->topic_filter = mqtt_malloc(strlen(topicFilter) + 1); if (NULL == handler->topic_filter) { mqtt_free(handler); return FAIL_RETURN; } memset((char *)handler->topic_filter, 0, strlen(topicFilter) + 1); #else if (strlen(topicFilter) >= CONFIG_MQTT_TOPIC_MAXLEN) { memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); return MQTT_TOPIC_LEN_TOO_SHORT; } memset((char *)handler->topic_filter, 0, CONFIG_MQTT_TOPIC_MAXLEN); #endif memcpy((char *)handler->topic_filter, topicFilter, strlen(topicFilter) + 1); #else if (strstr(topicFilter, "/+") != NULL || strstr(topicFilter, "/#") != NULL) { #ifdef PLATFORM_HAS_DYNMEM handler->topic_filter = mqtt_malloc(strlen(topicFilter) + 1); if (NULL == handler->topic_filter) { mqtt_free(handler); return FAIL_RETURN; } memset((char *)handler->topic_filter, 0, strlen(topicFilter) + 1); #else if (strlen(topicFilter) >= CONFIG_MQTT_TOPIC_MAXLEN) { memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); return MQTT_TOPIC_LEN_TOO_SHORT; } memset((char *)handler->topic_filter, 0, CONFIG_MQTT_TOPIC_MAXLEN); #endif handler->topic_type = TOPIC_FILTER_TYPE; memcpy((char *)handler->topic_filter, topicFilter, strlen(topicFilter) + 1); } else { #ifdef PLATFORM_HAS_DYNMEM handler->topic_filter = mqtt_malloc(MQTT_ZIP_PATH_DEFAULT_LEN); if (NULL == handler->topic_filter) { mqtt_free(handler); return FAIL_RETURN; } memset((char *)handler->topic_filter, 0, MQTT_ZIP_PATH_DEFAULT_LEN); #else if (MQTT_ZIP_PATH_DEFAULT_LEN >= CONFIG_MQTT_TOPIC_MAXLEN) { memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); return MQTT_TOPIC_LEN_TOO_SHORT; } memset((char *)handler->topic_filter, 0, CONFIG_MQTT_TOPIC_MAXLEN); #endif handler->topic_type = TOPIC_NAME_TYPE; if (iotx_mc_get_zip_topic(topicFilter, strlen(topicFilter), (char *)handler->topic_filter, MQTT_ZIP_PATH_DEFAULT_LEN) != 0) { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #else memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #endif return FAIL_RETURN; } } #endif handler->handle.h_fp = messageHandler; handler->handle.pcontext = pcontext; #ifdef SUB_PERSISTENCE_ENABLED if (qos == IOTX_MQTT_QOS3_SUB_LOCAL) { uint8_t dup = 0; #ifdef PLATFORM_HAS_DYNMEM iotx_mc_topic_handle_t *node; #endif HAL_MutexLock(c->lock_generic); #ifdef PLATFORM_HAS_DYNMEM #if defined(INSPECT_MQTT_FLOW) && defined (INFRA_LOG) #if WITH_MQTT_ZIP_TOPIC HEXDUMP_DEBUG(handler->topic_filter, MQTT_ZIP_PATH_DEFAULT_LEN); #else mqtt_warning("handler->topic: %s", handler->topic_filter); #endif #endif linkkit_list_for_each_entry(node, &c->list_sub_handle, linked_list, iotx_mc_topic_handle_t) { /* If subscribe the same topic and callback function, then ignore */ #if defined(INSPECT_MQTT_FLOW) && defined (INFRA_LOG) #if WITH_MQTT_ZIP_TOPIC HEXDUMP_DEBUG(node->topic_filter, MQTT_ZIP_PATH_DEFAULT_LEN); #else mqtt_warning("node->topic: %s", node->topic_filter); #endif #endif if (0 == iotx_mc_check_handle_is_identical(node, handler)) { mqtt_warning("dup sub,topic = %s", topicFilter); dup = 1; } } #else for (idx = 0; idx < IOTX_MC_SUBHANDLE_LIST_MAX_LEN; idx++) { /* If subscribe the same topic and callback function, then ignore */ if (&c->list_sub_handle[idx] != handler && 0 == iotx_mc_check_handle_is_identical(&c->list_sub_handle[idx], handler)) { mqtt_warning("dup sub,topic = %s", topicFilter); dup = 1; } } #endif if (dup == 0) { #ifdef PLATFORM_HAS_DYNMEM list_add_tail(&handler->linked_list, &c->list_sub_handle); #endif } else { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #else memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #endif } HAL_MutexUnlock(c->lock_generic); return SUCCESS_RETURN; } #endif HAL_MutexLock(c->lock_write_buf); if (_alloc_send_buffer(c, strlen(topicFilter)) < 0) { HAL_MutexUnlock(c->lock_write_buf); #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #else memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #endif return FAIL_RETURN; } len = MQTTSerialize_subscribe((unsigned char *)c->buf_send, c->buf_size_send, 0, (unsigned short)msgId, 1, &topic, (int *)&qos); if (len <= 0) { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #else memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #endif _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return MQTT_SUBSCRIBE_PACKET_ERROR; } mqtt_debug("%20s : %08d", "Packet Ident", msgId); mqtt_debug("%20s : %s", "Topic", topicFilter); mqtt_debug("%20s : %d", "QoS", (int)qos); mqtt_debug("%20s : %d", "Packet Length", len); #if defined(INSPECT_MQTT_FLOW) && defined (INFRA_LOG) HEXDUMP_DEBUG(c->buf_send, len); #endif if ((iotx_mc_send_packet(c, c->buf_send, len, &timer)) != SUCCESS_RETURN) { /* send the subscribe packet */ /* If send failed, remove it */ mqtt_err("run sendPacket error!"); #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #else memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #endif _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return MQTT_NETWORK_ERROR; } _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); { uint8_t dup = 0; #ifdef PLATFORM_HAS_DYNMEM iotx_mc_topic_handle_t *node; #endif HAL_MutexLock(c->lock_generic); #ifdef PLATFORM_HAS_DYNMEM #if defined(INSPECT_MQTT_FLOW) && defined (INFRA_LOG) #if WITH_MQTT_ZIP_TOPIC HEXDUMP_DEBUG(handler->topic_filter, MQTT_ZIP_PATH_DEFAULT_LEN); #else mqtt_warning("handler->topic: %s", handler->topic_filter); #endif #endif linkkit_list_for_each_entry(node, &c->list_sub_handle, linked_list, iotx_mc_topic_handle_t) { /* If subscribe the same topic and callback function, then ignore */ #if defined(INSPECT_MQTT_FLOW) && defined (INFRA_LOG) #if WITH_MQTT_ZIP_TOPIC HEXDUMP_DEBUG(node->topic_filter, MQTT_ZIP_PATH_DEFAULT_LEN); #else mqtt_warning("node->topic: %s", node->topic_filter); #endif #endif if (0 == iotx_mc_check_handle_is_identical(node, handler)) { mqtt_warning("dup sub,topic = %s", topicFilter); dup = 1; } } #else for (idx = 0; idx < IOTX_MC_SUBHANDLE_LIST_MAX_LEN; idx++) { /* If subscribe the same topic and callback function, then ignore */ if (&c->list_sub_handle[idx] != handler && 0 == iotx_mc_check_handle_is_identical(&c->list_sub_handle[idx], handler)) { mqtt_warning("dup sub,topic = %s", topicFilter); dup = 1; } } #endif if (dup == 0) { #ifdef PLATFORM_HAS_DYNMEM list_add_tail(&handler->linked_list, &c->list_sub_handle); #endif } else { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #else memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #endif } HAL_MutexUnlock(c->lock_generic); } return SUCCESS_RETURN; } static int iotx_mc_get_next_packetid(iotx_mc_client_t *c) { unsigned int id = 0; if (!c) { return FAIL_RETURN; } HAL_MutexLock(c->lock_generic); c->packet_id = (c->packet_id == IOTX_MC_PACKET_ID_MAX) ? 1 : c->packet_id + 1; id = c->packet_id; HAL_MutexUnlock(c->lock_generic); return id; } static int iotx_mc_check_rule(char *iterm, iotx_mc_topic_type_t type) { int i = 0; int len = 0; if (NULL == iterm) { mqtt_err("iterm is NULL"); return FAIL_RETURN; } len = strlen(iterm); for (i = 0; i < len; i++) { if (TOPIC_FILTER_TYPE == type) { if ('+' == iterm[i] || '#' == iterm[i]) { if (1 != len) { mqtt_err("the character # and + is error"); return FAIL_RETURN; } } } else { if ('+' == iterm[i] || '#' == iterm[i]) { mqtt_err("has character # and + is error"); return FAIL_RETURN; } } if (iterm[i] < 32 || iterm[i] >= 127) { return FAIL_RETURN; } } return SUCCESS_RETURN; } static int iotx_mc_check_topic(const char *topicName, iotx_mc_topic_type_t type) { int mask = 0; char *delim = "/"; char *iterm = NULL; char topicString[CONFIG_MQTT_TOPIC_MAXLEN]; if (NULL == topicName || '/' != topicName[0]) { return FAIL_RETURN; } if (strlen(topicName) > CONFIG_MQTT_TOPIC_MAXLEN) { mqtt_err("len of topicName exceeds %d", CONFIG_MQTT_TOPIC_MAXLEN); return FAIL_RETURN; } memset(topicString, 0x0, CONFIG_MQTT_TOPIC_MAXLEN); strncpy(topicString, topicName, CONFIG_MQTT_TOPIC_MAXLEN - 1); iterm = infra_strtok(topicString, delim); if (SUCCESS_RETURN != iotx_mc_check_rule(iterm, type)) { mqtt_err("run iotx_check_rule error"); return FAIL_RETURN; } for (;;) { iterm = infra_strtok(NULL, delim); if (iterm == NULL) { break; } /* The character '#' is not in the last */ if (1 == mask) { mqtt_err("the character # is error"); return FAIL_RETURN; } if (SUCCESS_RETURN != iotx_mc_check_rule(iterm, type)) { mqtt_err("run iotx_check_rule error"); return FAIL_RETURN; } if (iterm[0] == '#') { mask = 1; } } return SUCCESS_RETURN; } static inline int _is_in_yield_cb() { return _in_yield_cb; } static int MQTTUnsubscribe(iotx_mc_client_t *c, const char *topicFilter, unsigned int msgId) { MQTTString cur_topic; iotx_time_t timer; MQTTString topic = MQTTString_initializer; int len = 0; /*iotx_mc_topic_handle_t handler = {topicFilter, {NULL, NULL}};*/ iotx_mc_topic_handle_t *handler = NULL; #ifdef PLATFORM_HAS_DYNMEM iotx_mc_topic_handle_t *node = NULL; iotx_mc_topic_handle_t *next = NULL; #else int idx = 0; iotx_mc_topic_handle_t s_handler; #endif if (!c || !topicFilter) { return FAIL_RETURN; } topic.cstring = (char *)topicFilter; iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); #ifdef PLATFORM_HAS_DYNMEM handler = mqtt_malloc(sizeof(iotx_mc_topic_handle_t)); if (NULL == handler) { return FAIL_RETURN; } #else handler = &s_handler; #endif memset(handler, 0, sizeof(iotx_mc_topic_handle_t)); #if !(WITH_MQTT_ZIP_TOPIC) #ifdef PLATFORM_HAS_DYNMEM handler->topic_filter = mqtt_malloc(strlen(topicFilter) + 1); if (NULL == handler->topic_filter) { mqtt_free(handler); return FAIL_RETURN; } memset((char *)handler->topic_filter, 0, strlen(topicFilter) + 1); #else if (strlen(topicFilter) >= CONFIG_MQTT_TOPIC_MAXLEN) { return MQTT_TOPIC_LEN_TOO_SHORT; } memset((char *)handler->topic_filter, 0, CONFIG_MQTT_TOPIC_MAXLEN); #endif memcpy((char *)handler->topic_filter, topicFilter, strlen(topicFilter) + 1); #else if (strstr(topicFilter, "/+") != NULL || strstr(topicFilter, "/#") != NULL) { #ifdef PLATFORM_HAS_DYNMEM handler->topic_filter = mqtt_malloc(strlen(topicFilter) + 1); if (NULL == handler->topic_filter) { mqtt_free(handler); return FAIL_RETURN; } memset((char *)handler->topic_filter, 0, strlen(topicFilter) + 1); #else if (strlen(topicFilter) >= CONFIG_MQTT_TOPIC_MAXLEN) { return MQTT_TOPIC_LEN_TOO_SHORT; } memset((char *)handler->topic_filter, 0, CONFIG_MQTT_TOPIC_MAXLEN); #endif handler->topic_type = TOPIC_FILTER_TYPE; memcpy((char *)handler->topic_filter, topicFilter, strlen(topicFilter) + 1); } else { #ifdef PLATFORM_HAS_DYNMEM handler->topic_filter = mqtt_malloc(MQTT_ZIP_PATH_DEFAULT_LEN); if (NULL == handler->topic_filter) { mqtt_free(handler); return FAIL_RETURN; } memset((char *)handler->topic_filter, 0, MQTT_ZIP_PATH_DEFAULT_LEN); #else if (MQTT_ZIP_PATH_DEFAULT_LEN >= CONFIG_MQTT_TOPIC_MAXLEN) { return MQTT_TOPIC_LEN_TOO_SHORT; } memset((char *)handler->topic_filter, 0, CONFIG_MQTT_TOPIC_MAXLEN); #endif handler->topic_type = TOPIC_NAME_TYPE; if (iotx_mc_get_zip_topic(topicFilter, strlen(topicFilter), (char *)handler->topic_filter, MQTT_ZIP_PATH_DEFAULT_LEN) != 0) { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #endif return FAIL_RETURN; } } #endif HAL_MutexLock(c->lock_write_buf); if (_alloc_send_buffer(c, strlen(topicFilter)) < 0) { HAL_MutexUnlock(c->lock_write_buf); #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #endif return FAIL_RETURN; } if ((len = MQTTSerialize_unsubscribe((unsigned char *)c->buf_send, c->buf_size_send, 0, (unsigned short)msgId, 1, &topic)) <= 0) { #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #endif _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return MQTT_UNSUBSCRIBE_PACKET_ERROR; } if ((iotx_mc_send_packet(c, c->buf_send, len, &timer)) != SUCCESS_RETURN) { /* send the subscribe packet */ #ifdef PLATFORM_HAS_DYNMEM mqtt_free(handler->topic_filter); mqtt_free(handler); #endif _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return MQTT_NETWORK_ERROR; } cur_topic.cstring = NULL; cur_topic.lenstring.data = (char *)handler->topic_filter; #if !(WITH_MQTT_ZIP_TOPIC) cur_topic.lenstring.len = strlen(handler->topic_filter) + 1; #else if (handler->topic_type == TOPIC_FILTER_TYPE) { cur_topic.lenstring.len = strlen(handler->topic_filter) + 1; } else { cur_topic.lenstring.len = MQTT_ZIP_PATH_DEFAULT_LEN; } #endif /* we have to find the right message handler - indexed by topic */ HAL_MutexLock(c->lock_generic); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry_safe(node, next, &c->list_sub_handle, linked_list, iotx_mc_topic_handle_t) { if (MQTTPacket_equals(&cur_topic, (char *)node->topic_filter) || iotx_mc_is_topic_matched((char *)node->topic_filter, &cur_topic)) { mqtt_debug("topic be matched"); list_del(&node->linked_list); mqtt_free(node->topic_filter); mqtt_free(node); } } mqtt_free(handler->topic_filter); mqtt_free(handler); #else for (idx = 0; idx < IOTX_MC_SUBHANDLE_LIST_MAX_LEN; idx++) { if ((c->list_sub_handle[idx].used == 1) && (MQTTPacket_equals(&cur_topic, (char *)c->list_sub_handle[idx].topic_filter) || iotx_mc_is_topic_matched((char *)c->list_sub_handle[idx].topic_filter, &cur_topic))) { mqtt_debug("topic be matched"); memset(&c->list_sub_handle[idx], 0, sizeof(iotx_mc_topic_handle_t)); } } #endif HAL_MutexUnlock(c->lock_generic); _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return SUCCESS_RETURN; } int MQTTPublish(iotx_mc_client_t *c, const char *topicName, iotx_mqtt_topic_info_pt topic_msg) { iotx_time_t timer; MQTTString topic = MQTTString_initializer; int len = 0; #if !WITH_MQTT_ONLY_QOS0 iotx_mc_pub_info_t *node = NULL; #endif #ifdef INFRA_LOG_NETWORK_PAYLOAD const char *json_payload = NULL; #endif if (!c || !topicName || !topic_msg) { return FAIL_RETURN; } topic.cstring = (char *)topicName; iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); HAL_MutexLock(c->lock_list_pub); HAL_MutexLock(c->lock_write_buf); if (_alloc_send_buffer(c, strlen(topicName) + topic_msg->payload_len) < 0) { HAL_MutexUnlock(c->lock_write_buf); HAL_MutexUnlock(c->lock_list_pub); return FAIL_RETURN; } len = MQTTSerialize_publish((unsigned char *)c->buf_send, c->buf_size_send, 0, topic_msg->qos, topic_msg->retain, topic_msg->packet_id, topic, (unsigned char *)topic_msg->payload, topic_msg->payload_len); if (len <= 0) { mqtt_err("MQTTSerialize_publish is error, len=%d, buf_size_send=%u, payloadlen=%u", len, c->buf_size_send, topic_msg->payload_len); _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); HAL_MutexUnlock(c->lock_list_pub); return MQTT_PUBLISH_PACKET_ERROR; } #if !WITH_MQTT_ONLY_QOS0 node = NULL; /* If the QOS >1, push the information into list of wait publish ACK */ if (topic_msg->qos > IOTX_MQTT_QOS0) { /* push into list */ if (SUCCESS_RETURN != iotx_mc_push_pubInfo_to(c, len, topic_msg->packet_id, &node)) { mqtt_err("push publish into to pubInfolist failed!"); _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); HAL_MutexUnlock(c->lock_list_pub); return MQTT_PUSH_TO_LIST_ERROR; } } #endif /* send the publish packet */ if (iotx_mc_send_packet(c, c->buf_send, len, &timer) != SUCCESS_RETURN) { #if !WITH_MQTT_ONLY_QOS0 if (topic_msg->qos > IOTX_MQTT_QOS0) { /* If not even successfully sent to IP stack, meaningless to wait QOS1 ack, give up waiting */ #ifdef PLATFORM_HAS_DYNMEM list_del(&node->linked_list); mqtt_free(node); #else memset(node, 0, sizeof(iotx_mc_pub_info_t)); #endif } #endif _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); HAL_MutexUnlock(c->lock_list_pub); return MQTT_NETWORK_ERROR; } #ifdef INFRA_LOG_NETWORK_PAYLOAD json_payload = (const char *)topic_msg->payload; mqtt_info("Upstream Topic: '%s'", topicName); mqtt_info("Upstream Payload:"); iotx_facility_json_print(json_payload, LOG_INFO_LEVEL, '>'); #endif /* #ifdef INFRA_LOG */ _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); HAL_MutexUnlock(c->lock_list_pub); return SUCCESS_RETURN; } static int MQTTDisconnect(iotx_mc_client_t *c) { int rc = FAIL_RETURN; int len = 0; iotx_time_t timer; /* we might wait for incomplete incoming publishes to complete */ if (!c) { return FAIL_RETURN; } HAL_MutexLock(c->lock_write_buf); if (_alloc_send_buffer(c, 0) < 0) { HAL_MutexUnlock(c->lock_write_buf); return FAIL_RETURN; } len = MQTTSerialize_disconnect((unsigned char *)c->buf_send, c->buf_size_send); iotx_time_init(&timer); utils_time_countdown_ms(&timer, c->request_timeout_ms); if (len > 0) { rc = iotx_mc_send_packet(c, c->buf_send, len, &timer); /* send the disconnect packet */ } _reset_send_buffer(c); HAL_MutexUnlock(c->lock_write_buf); return rc; } static int iotx_mc_disconnect(iotx_mc_client_t *pClient) { int rc = -1; if (NULL == pClient) { return NULL_VALUE_ERROR; } if (wrapper_mqtt_check_state(pClient)) { rc = MQTTDisconnect(pClient); mqtt_debug("rc = MQTTDisconnect() = %d", rc); rc = rc; } /* close tcp/ip socket or free tls resources */ pClient->ipstack.disconnect(&pClient->ipstack); iotx_mc_set_client_state(pClient, IOTX_MC_STATE_INITIALIZED); mqtt_info("mqtt disconnect!"); return SUCCESS_RETURN; } /************************ Public Interface ************************/ void *DY_wrapper_mqtt_init(iotx_mqtt_param_t *mqtt_params, char* device_name, char* product_key) { int err; iotx_mc_client_t *pclient = NULL; #ifndef PLATFORM_HAS_DYNMEM int idx; #endif #ifdef PLATFORM_HAS_DYNMEM pclient = (iotx_mc_client_t *)mqtt_malloc(sizeof(iotx_mc_client_t)); if (NULL == pclient) { mqtt_err("not enough memory."); return NULL; } memset(pclient, 0, sizeof(iotx_mc_client_t)); #else for (idx = 0; idx < IOTX_MC_CLIENT_MAX_COUNT; idx++) { if (g_iotx_mc_client[idx].used == 0) { g_iotx_mc_client[idx].used = 1; pclient = &g_iotx_mc_client[idx]; break; } } if (NULL == pclient) { mqtt_err("IOTX_MC_CLIENT_MAX_COUNT too short: %d", IOTX_MC_CLIENT_MAX_COUNT); return NULL; } #endif err = iotx_mc_init(pclient, mqtt_params); if (SUCCESS_RETURN != err) { mqtt_err("iotx_mc_init failed"); iotx_mc_release(pclient); return NULL; } return pclient; } /************************ Public Interface ************************/ void *wrapper_mqtt_init(iotx_mqtt_param_t *mqtt_params) { int err; iotx_mc_client_t *pclient = NULL; #ifndef PLATFORM_HAS_DYNMEM int idx; #endif #ifdef PLATFORM_HAS_DYNMEM pclient = (iotx_mc_client_t *)mqtt_malloc(sizeof(iotx_mc_client_t)); if (NULL == pclient) { mqtt_err("not enough memory."); return NULL; } memset(pclient, 0, sizeof(iotx_mc_client_t)); #else for (idx = 0; idx < IOTX_MC_CLIENT_MAX_COUNT; idx++) { if (g_iotx_mc_client[idx].used == 0) { g_iotx_mc_client[idx].used = 1; pclient = &g_iotx_mc_client[idx]; break; } } if (NULL == pclient) { mqtt_err("IOTX_MC_CLIENT_MAX_COUNT too short: %d", IOTX_MC_CLIENT_MAX_COUNT); return NULL; } #endif err = iotx_mc_init(pclient, mqtt_params); if (SUCCESS_RETURN != err) { mqtt_err("iotx_mc_init failed"); iotx_mc_release(pclient); return NULL; } return pclient; } int wrapper_mqtt_connect(void *client) { int rc = FAIL_RETURN; int retry_max = 3; int retry_cnt = 1; int retry_interval = 1000; iotx_mc_client_t *pClient = (iotx_mc_client_t *)client; if (NULL == pClient) { return NULL_VALUE_ERROR; } /* Establish TCP or TLS connection */ do { mqtt_debug("calling TCP or TLS connect HAL for [%d/%d] iteration", retry_cnt, retry_max); rc = pClient->ipstack.connect(&pClient->ipstack); if (SUCCESS_RETURN != rc) { pClient->ipstack.disconnect(&pClient->ipstack); mqtt_err("TCP or TLS Connection failed"); if (ERROR_CERTIFICATE_EXPIRED == rc) { mqtt_err("certificate is expired! rc = %d", rc); rc = ERROR_CERT_VERIFY_FAIL; HAL_SleepMs(retry_interval); continue; } else { rc = MQTT_NETWORK_CONNECT_ERROR; HAL_SleepMs(retry_interval); continue; } } else { mqtt_debug("rc = pClient->ipstack.connect() = %d, success @ [%d/%d] iteration", rc, retry_cnt, retry_max); break; } } while (++retry_cnt <= retry_max); #ifdef ASYNC_PROTOCOL_STACK iotx_mc_set_client_state(pClient, IOTX_MC_STATE_CONNECT_BLOCK); rc = MQTT_CONNECT_BLOCK; #else rc = _mqtt_connect(pClient); #endif return rc; } int wrapper_mqtt_release(void **c) { iotx_mc_client_t *pClient; #ifdef PLATFORM_HAS_DYNMEM iotx_mc_topic_handle_t *node = NULL, *next = NULL; #endif if (NULL == c) { return NULL_VALUE_ERROR; } pClient = (iotx_mc_client_t *)*c; if (NULL == pClient) { return NULL_VALUE_ERROR; } /* iotx_delete_thread(pClient); */ HAL_SleepMs(100); iotx_mc_disconnect(pClient); iotx_mc_set_client_state(pClient, IOTX_MC_STATE_INVALID); HAL_SleepMs(100); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry_safe(node, next, &pClient->list_sub_handle, linked_list, iotx_mc_topic_handle_t) { list_del(&node->linked_list); mqtt_free(node->topic_filter); mqtt_free(node); } #else memset(pClient->list_sub_handle, 0, sizeof(iotx_mc_topic_handle_t) * IOTX_MC_SUBHANDLE_LIST_MAX_LEN); #endif HAL_MutexDestroy(pClient->lock_generic); HAL_MutexDestroy(pClient->lock_list_pub); HAL_MutexDestroy(pClient->lock_write_buf); HAL_MutexDestroy(pClient->lock_yield); HAL_MutexDestroy(pClient->lock_read_buf); #if !WITH_MQTT_ONLY_QOS0 iotx_mc_pub_wait_list_deinit(pClient); #endif #ifdef PLATFORM_HAS_DYNMEM if (pClient->buf_send != NULL) { mqtt_free(pClient->buf_send); pClient->buf_send = NULL; } if (pClient->buf_read != NULL) { mqtt_free(pClient->buf_read); pClient->buf_read = NULL; } mqtt_free(pClient); #else memset(pClient, 0, sizeof(iotx_mc_client_t)); #endif *c = NULL; mqtt_info("mqtt release!"); return SUCCESS_RETURN; } int wrapper_mqtt_yield(void *client, int timeout_ms) { iotx_mc_client_t *pClient = (iotx_mc_client_t *)client; if (pClient == NULL) { return NULL_VALUE_ERROR; } if (timeout_ms < 0) { mqtt_err("Invalid argument, timeout_ms = %d", timeout_ms); return -1; } if (timeout_ms == 0) { timeout_ms = 10; } HAL_MutexLock(pClient->lock_yield); pClient->cycle_timeout_ms = timeout_ms; /* Keep MQTT alive or reconnect if connection abort */ iotx_mc_keepalive(pClient); HAL_MutexUnlock(pClient->lock_yield); #ifndef ASYNC_PROTOCOL_STACK _mqtt_cycle(client); #else if (pClient->client_state == IOTX_MC_STATE_CONNECTED) { #if !WITH_MQTT_ONLY_QOS0 /* check list of wait publish ACK to remove node that is ACKED or timeout */ MQTTPubInfoProc(pClient); #endif } HAL_SleepMs(timeout_ms); #endif return 0; } /* check MQTT client is in normal state */ /* 0, in abnormal state; 1, in normal state */ int wrapper_mqtt_check_state(void *client) { if (!client) { return 0; } if (iotx_mc_get_client_state((iotx_mc_client_t *)client) == IOTX_MC_STATE_CONNECTED) { return 1; } return 0; } int wrapper_mqtt_subscribe(void *client, const char *topicFilter, iotx_mqtt_qos_t qos, iotx_mqtt_event_handle_func_fpt topic_handle_func, void *pcontext) { int rc = FAIL_RETURN; unsigned int msgId; iotx_mc_client_t *c; if (NULL == client || NULL == topicFilter || strlen(topicFilter) == 0 || !topic_handle_func) { mqtt_err(" paras error"); return NULL_VALUE_ERROR; } c = (iotx_mc_client_t *)client; msgId = iotx_mc_get_next_packetid(c); if (!wrapper_mqtt_check_state(c)) { mqtt_err("mqtt client state is error,state = %d", iotx_mc_get_client_state(c)); return MQTT_STATE_ERROR; } if (0 != iotx_mc_check_topic(topicFilter, TOPIC_FILTER_TYPE)) { mqtt_err("topic format is error,topicFilter = %s", topicFilter); return MQTT_TOPIC_FORMAT_ERROR; } mqtt_debug("PERFORM subscribe to '%s' (msgId=%d)", topicFilter, msgId); rc = MQTTSubscribe(c, topicFilter, qos, msgId, topic_handle_func, pcontext); if (rc != SUCCESS_RETURN) { if (rc == MQTT_NETWORK_ERROR) { iotx_mc_set_client_state(c, IOTX_MC_STATE_DISCONNECTED); } mqtt_err("run MQTTSubscribe error, rc = %d", rc); return rc; } mqtt_info("mqtt subscribe packet sent,topic = %s!", topicFilter); return msgId; } int wrapper_mqtt_subscribe_sync(void *c, const char *topic_filter, iotx_mqtt_qos_t qos, iotx_mqtt_event_handle_func_fpt topic_handle_func, void *pcontext, int timeout_ms) { int subed; int ret; iotx_time_t timer; iotx_mc_client_t *client = (iotx_mc_client_t *)c; int cnt = 0; mqtt_sub_sync_node_t *node = NULL; #ifdef PLATFORM_HAS_DYNMEM mqtt_sub_sync_node_t *next = NULL; #else int idx = 0; #endif if (client == NULL) { return NULL_VALUE_ERROR; } #ifdef SUB_PERSISTENCE_ENABLED if (qos > IOTX_MQTT_QOS3_SUB_LOCAL) { mqtt_warning("Invalid qos(%d) out of [%d, %d], using %d", qos, IOTX_MQTT_QOS0, IOTX_MQTT_QOS3_SUB_LOCAL, IOTX_MQTT_QOS0); qos = IOTX_MQTT_QOS0; } #else if (qos > IOTX_MQTT_QOS2) { mqtt_warning("Invalid qos(%d) out of [%d, %d], using %d", qos, IOTX_MQTT_QOS0, IOTX_MQTT_QOS2, IOTX_MQTT_QOS0); qos = IOTX_MQTT_QOS0; } #endif iotx_time_init(&timer); utils_time_countdown_ms(&timer, timeout_ms); ret = -1; subed = 0; cnt = 0; cnt = cnt; do { #ifdef PLATFORM_HAS_DYNMEM mqtt_sub_sync_node_t *node = NULL; mqtt_sub_sync_node_t *next = NULL; #else int idx = 0; #endif if (ret < 0) { ret = wrapper_mqtt_subscribe(client, topic_filter, qos, topic_handle_func, pcontext); if (_is_in_yield_cb() != 0 || qos == IOTX_MQTT_QOS3_SUB_LOCAL) { return ret; } } if (!subed && ret >= 0) { mqtt_sub_sync_node_t *node = NULL; #ifndef PLATFORM_HAS_DYNMEM int idx = 0; #endif #ifdef PLATFORM_HAS_DYNMEM node = (mqtt_sub_sync_node_t *)mqtt_malloc(sizeof(mqtt_sub_sync_node_t)); #else for (idx = 0; idx < IOTX_MC_SUBSYNC_LIST_MAX_LEN; idx++) { if (client->list_sub_sync_ack[idx].used == 0) { memset(&client->list_sub_sync_ack[idx], 0, sizeof(mqtt_sub_sync_node_t)); client->list_sub_sync_ack[idx].used = 1; node = &client->list_sub_sync_ack[idx]; break; } } #endif if (node != NULL) { mqtt_debug("packet_id = %d", ret); node->packet_id = ret; node->ack_type = IOTX_MQTT_EVENT_UNDEF; #ifdef PLATFORM_HAS_DYNMEM HAL_MutexLock(client->lock_generic); list_add_tail(&node->linked_list, &client->list_sub_sync_ack); HAL_MutexUnlock(client->lock_generic); #endif subed = 1; } } wrapper_mqtt_yield(client, 100); HAL_MutexLock(client->lock_generic); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry_safe(node, next, &client->list_sub_sync_ack, linked_list, mqtt_sub_sync_node_t) { if (node->packet_id == ret) { mqtt_debug("node->ack_type=%d cnt=%d", node->ack_type, cnt++); if (node->ack_type == IOTX_MQTT_EVENT_SUBCRIBE_SUCCESS) { list_del(&node->linked_list); mqtt_free(node); mqtt_debug("success!!"); HAL_MutexUnlock(client->lock_generic); return ret; } else if (node->ack_type == IOTX_MQTT_EVENT_SUBCRIBE_NACK) { list_del(&node->linked_list); mqtt_free(node); ret = -1; /* resub */ subed = 0; } else if (node->ack_type == IOTX_MQTT_EVENT_SUBCRIBE_TIMEOUT) { list_del(&node->linked_list); mqtt_free(node); ret = -1; /* resub */ subed = 0; } } break; } #else for (idx = 0; idx < IOTX_MC_SUBSYNC_LIST_MAX_LEN; idx++) { if (client->list_sub_sync_ack[idx].used == 0) { continue; } if (client->list_sub_sync_ack[idx].packet_id == ret) { mqtt_debug("client->list_sub_sync_ack[%d].ack_type=%d cnt=%d", idx, client->list_sub_sync_ack[idx].ack_type, cnt++); if (client->list_sub_sync_ack[idx].ack_type == IOTX_MQTT_EVENT_SUBCRIBE_SUCCESS) { memset(&client->list_sub_sync_ack[idx], 0, sizeof(mqtt_sub_sync_node_t)); mqtt_debug("success!!"); HAL_MutexUnlock(client->lock_generic); return ret; } else if (client->list_sub_sync_ack[idx].ack_type == IOTX_MQTT_EVENT_SUBCRIBE_NACK) { memset(&client->list_sub_sync_ack[idx], 0, sizeof(mqtt_sub_sync_node_t)); ret = -1; /* resub */ subed = 0; } else if (client->list_sub_sync_ack[idx].ack_type == IOTX_MQTT_EVENT_SUBCRIBE_TIMEOUT) { memset(&client->list_sub_sync_ack[idx], 0, sizeof(mqtt_sub_sync_node_t)); ret = -1; /* resub */ subed = 0; } } break; } #endif HAL_MutexUnlock(client->lock_generic); } while (!utils_time_is_expired(&timer)); mqtt_warning("sync subscribe time out!!"); HAL_MutexLock(client->lock_generic); #ifdef PLATFORM_HAS_DYNMEM linkkit_list_for_each_entry_safe(node, next, &client->list_sub_sync_ack, linked_list, mqtt_sub_sync_node_t) { if (node->packet_id == ret) { list_del(&node->linked_list); mqtt_free(node); } } #else for (idx = 0; idx < IOTX_MC_SUBSYNC_LIST_MAX_LEN; idx++) { if (client->list_sub_sync_ack[idx].used && node && node->packet_id == ret) { memset(&client->list_sub_sync_ack[idx], 0, sizeof(mqtt_sub_sync_node_t)); } } #endif HAL_MutexUnlock(client->lock_generic); return -1; } int wrapper_mqtt_unsubscribe(void *client, const char *topicFilter) { int rc = FAIL_RETURN; iotx_mc_client_t *c = (iotx_mc_client_t *)client; unsigned int msgId; if (NULL == c || NULL == topicFilter) { return NULL_VALUE_ERROR; } msgId = iotx_mc_get_next_packetid(c); if (0 != iotx_mc_check_topic(topicFilter, TOPIC_FILTER_TYPE)) { mqtt_err("topic format is error,topicFilter = %s", topicFilter); return MQTT_TOPIC_FORMAT_ERROR; } if (!wrapper_mqtt_check_state(c)) { mqtt_err("mqtt client state is error,state = %d", iotx_mc_get_client_state(c)); return MQTT_STATE_ERROR; } rc = MQTTUnsubscribe(c, topicFilter, msgId); if (rc != SUCCESS_RETURN) { if (rc == MQTT_NETWORK_ERROR) { /* send the subscribe packet */ iotx_mc_set_client_state(c, IOTX_MC_STATE_DISCONNECTED); } mqtt_err("run MQTTUnsubscribe error!, rc = %d", rc); return rc; } mqtt_info("mqtt unsubscribe packet sent,topic = %s!", topicFilter); return (int)msgId; } int wrapper_mqtt_publish(void *client, const char *topicName, iotx_mqtt_topic_info_pt topic_msg) { uint16_t msg_id = 0; int rc = FAIL_RETURN; iotx_mc_client_t *c = (iotx_mc_client_t *)client; if (c == NULL || topicName == NULL || topic_msg == NULL || topic_msg->payload == NULL) { return NULL_VALUE_ERROR; } if (0 != iotx_mc_check_topic(topicName, TOPIC_NAME_TYPE)) { mqtt_err("topic format is error,topicFilter = %s", topicName); return MQTT_TOPIC_FORMAT_ERROR; } if (!wrapper_mqtt_check_state(c)) { mqtt_err("mqtt client state is error,state = %d", iotx_mc_get_client_state(c)); return MQTT_STATE_ERROR; } #if !WITH_MQTT_ONLY_QOS0 if (topic_msg->qos == IOTX_MQTT_QOS1 || topic_msg->qos == IOTX_MQTT_QOS2) { msg_id = iotx_mc_get_next_packetid(c); topic_msg->packet_id = msg_id; } if (topic_msg->qos == IOTX_MQTT_QOS2) { mqtt_err("MQTTPublish return error,MQTT_QOS2 is now not supported."); return MQTT_PUBLISH_QOS_ERROR; } #else topic_msg->qos = IOTX_MQTT_QOS0; #endif #if defined(INSPECT_MQTT_FLOW) && defined(INFRA_LOG) HEXDUMP_DEBUG(topicName, strlen(topicName)); HEXDUMP_DEBUG(topic_msg->payload, topic_msg->payload_len); #endif rc = MQTTPublish(c, topicName, topic_msg); if (rc != SUCCESS_RETURN) { /* send the subscribe packet */ if (rc == MQTT_NETWORK_ERROR) { iotx_mc_set_client_state(c, IOTX_MC_STATE_DISCONNECTED); } mqtt_err("MQTTPublish is error, rc = %d", rc); return rc; } return (int)msg_id; } #ifdef ASYNC_PROTOCOL_STACK int wrapper_mqtt_nwk_event_handler(void *client, iotx_mqtt_nwk_event_t event, iotx_mqtt_nwk_param_t *param) { int rc = FAIL_RETURN; iotx_mc_client_t *pClient = (iotx_mc_client_t *)client; if (client == NULL || event >= IOTX_MQTT_SOC_MAX) { return NULL_VALUE_ERROR; } switch (event) { case IOTX_MQTT_SOC_CONNECTED: { rc = _mqtt_connect(pClient); if (rc == SUCCESS_RETURN) { iotx_mc_set_client_state(pClient, IOTX_MC_STATE_CONNECTED); } } break; case IOTX_MQTT_SOC_CLOSE: { iotx_mc_set_client_state(pClient, IOTX_MC_STATE_DISCONNECTED); } break; case IOTX_MQTT_SOC_READ: { HAL_MutexLock(pClient->lock_yield); _mqtt_cycle(pClient); HAL_MutexUnlock(pClient->lock_yield); rc = SUCCESS_RETURN; } break; case IOTX_MQTT_SOC_WRITE: { } break; default: { mqtt_err("unknown event: %d", event); } break; } return rc; } #endif