kopia lustrzana https://github.com/martin-ger/esp_mqtt
Add debug info if platform don't support SSL due to some reason.
Add MQTT_ping.develop
Huang Rui
rodzic
d9e6e715d5
commit
c3d744ae94
257
mqtt/mqtt.c
257
mqtt/mqtt.c
|
|
@ -40,7 +40,7 @@
|
|||
#include "mqtt.h"
|
||||
#include "queue.h"
|
||||
|
||||
#define MQTT_TASK_PRIO 0
|
||||
#define MQTT_TASK_PRIO 2
|
||||
#define MQTT_TASK_QUEUE_SIZE 1
|
||||
#define MQTT_SEND_TIMOUT 5
|
||||
|
||||
|
|
@ -62,7 +62,7 @@ mqtt_dns_found(const char *name, ip_addr_t *ipaddr, void *arg)
|
|||
MQTT_Client* client = (MQTT_Client *)pConn->reverse;
|
||||
|
||||
|
||||
if(ipaddr == NULL)
|
||||
if (ipaddr == NULL)
|
||||
{
|
||||
INFO("DNS: Found, but got no ip, try to reconnect\r\n");
|
||||
client->connState = TCP_RECONNECT_REQ;
|
||||
|
|
@ -70,16 +70,20 @@ mqtt_dns_found(const char *name, ip_addr_t *ipaddr, void *arg)
|
|||
}
|
||||
|
||||
INFO("DNS: found ip %d.%d.%d.%d\n",
|
||||
*((uint8 *) &ipaddr->addr),
|
||||
*((uint8 *) &ipaddr->addr + 1),
|
||||
*((uint8 *) &ipaddr->addr + 2),
|
||||
*((uint8 *) &ipaddr->addr + 3));
|
||||
*((uint8 *) &ipaddr->addr),
|
||||
*((uint8 *) &ipaddr->addr + 1),
|
||||
*((uint8 *) &ipaddr->addr + 2),
|
||||
*((uint8 *) &ipaddr->addr + 3));
|
||||
|
||||
if(client->ip.addr == 0 && ipaddr->addr != 0)
|
||||
if (client->ip.addr == 0 && ipaddr->addr != 0)
|
||||
{
|
||||
os_memcpy(client->pCon->proto.tcp->remote_ip, &ipaddr->addr, 4);
|
||||
if(client->security){
|
||||
if (client->security) {
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_connect(client->pCon);
|
||||
#else
|
||||
INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
espconn_connect(client->pCon);
|
||||
|
|
@ -104,7 +108,7 @@ deliver_publish(MQTT_Client* client, uint8_t* message, int length)
|
|||
event_data.data_length = length;
|
||||
event_data.data = mqtt_get_publish_data(message, &event_data.data_length);
|
||||
|
||||
if(client->dataCb)
|
||||
if (client->dataCb)
|
||||
client->dataCb((uint32_t*)client, event_data.topic, event_data.topic_length, event_data.data, event_data.data_length);
|
||||
|
||||
}
|
||||
|
|
@ -129,19 +133,23 @@ mqtt_tcpclient_recv(void *arg, char *pdata, unsigned short len)
|
|||
|
||||
READPACKET:
|
||||
INFO("TCP: data received %d bytes\r\n", len);
|
||||
if(len < MQTT_BUF_SIZE && len > 0){
|
||||
if (len < MQTT_BUF_SIZE && len > 0) {
|
||||
os_memcpy(client->mqtt_state.in_buffer, pdata, len);
|
||||
|
||||
msg_type = mqtt_get_type(client->mqtt_state.in_buffer);
|
||||
msg_qos = mqtt_get_qos(client->mqtt_state.in_buffer);
|
||||
msg_id = mqtt_get_id(client->mqtt_state.in_buffer, client->mqtt_state.in_buffer_length);
|
||||
switch(client->connState){
|
||||
switch (client->connState) {
|
||||
case MQTT_CONNECT_SENDING:
|
||||
if(msg_type == MQTT_MSG_TYPE_CONNACK){
|
||||
if(client->mqtt_state.pending_msg_type != MQTT_MSG_TYPE_CONNECT){
|
||||
if (msg_type == MQTT_MSG_TYPE_CONNACK) {
|
||||
if (client->mqtt_state.pending_msg_type != MQTT_MSG_TYPE_CONNECT) {
|
||||
INFO("MQTT: Invalid packet\r\n");
|
||||
if(client->security){
|
||||
if (client->security) {
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_disconnect(client->pCon);
|
||||
#else
|
||||
INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
espconn_disconnect(client->pCon);
|
||||
|
|
@ -149,7 +157,7 @@ READPACKET:
|
|||
} else {
|
||||
INFO("MQTT: Connected to %s:%d\r\n", client->host, client->port);
|
||||
client->connState = MQTT_DATA;
|
||||
if(client->connectedCb)
|
||||
if (client->connectedCb)
|
||||
client->connectedCb((uint32_t*)client);
|
||||
}
|
||||
|
||||
|
|
@ -160,81 +168,81 @@ READPACKET:
|
|||
client->mqtt_state.message_length = mqtt_get_total_length(client->mqtt_state.in_buffer, client->mqtt_state.message_length_read);
|
||||
|
||||
|
||||
switch(msg_type)
|
||||
switch (msg_type)
|
||||
{
|
||||
|
||||
case MQTT_MSG_TYPE_SUBACK:
|
||||
if(client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_SUBSCRIBE && client->mqtt_state.pending_msg_id == msg_id)
|
||||
INFO("MQTT: Subscribe successful\r\n");
|
||||
case MQTT_MSG_TYPE_SUBACK:
|
||||
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_SUBSCRIBE && client->mqtt_state.pending_msg_id == msg_id)
|
||||
INFO("MQTT: Subscribe successful\r\n");
|
||||
break;
|
||||
case MQTT_MSG_TYPE_UNSUBACK:
|
||||
if(client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_UNSUBSCRIBE && client->mqtt_state.pending_msg_id == msg_id)
|
||||
INFO("MQTT: UnSubscribe successful\r\n");
|
||||
case MQTT_MSG_TYPE_UNSUBACK:
|
||||
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_UNSUBSCRIBE && client->mqtt_state.pending_msg_id == msg_id)
|
||||
INFO("MQTT: UnSubscribe successful\r\n");
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PUBLISH:
|
||||
if(msg_qos == 1)
|
||||
case MQTT_MSG_TYPE_PUBLISH:
|
||||
if (msg_qos == 1)
|
||||
client->mqtt_state.outbound_message = mqtt_msg_puback(&client->mqtt_state.mqtt_connection, msg_id);
|
||||
else if(msg_qos == 2)
|
||||
else if (msg_qos == 2)
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
|
||||
if(msg_qos == 1 || msg_qos == 2){
|
||||
if (msg_qos == 1 || msg_qos == 2) {
|
||||
INFO("MQTT: Queue response QoS: %d\r\n", msg_qos);
|
||||
if(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){
|
||||
if (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
}
|
||||
}
|
||||
|
||||
deliver_publish(client, client->mqtt_state.in_buffer, client->mqtt_state.message_length_read);
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PUBACK:
|
||||
if(client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_msg_id == msg_id){
|
||||
INFO("MQTT: received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish\r\n");
|
||||
case MQTT_MSG_TYPE_PUBACK:
|
||||
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_msg_id == msg_id) {
|
||||
INFO("MQTT: received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish\r\n");
|
||||
}
|
||||
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PUBREC:
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
|
||||
if(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
}
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PUBREL:
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
|
||||
if(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){
|
||||
case MQTT_MSG_TYPE_PUBREC:
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pubrel(&client->mqtt_state.mqtt_connection, msg_id);
|
||||
if (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
}
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PUBCOMP:
|
||||
if(client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_msg_id == msg_id){
|
||||
INFO("MQTT: receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish\r\n");
|
||||
}
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PINGREQ:
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pingresp(&client->mqtt_state.mqtt_connection);
|
||||
if(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){
|
||||
case MQTT_MSG_TYPE_PUBREL:
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pubcomp(&client->mqtt_state.mqtt_connection, msg_id);
|
||||
if (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
}
|
||||
}
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PINGRESP:
|
||||
case MQTT_MSG_TYPE_PUBCOMP:
|
||||
if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_msg_id == msg_id) {
|
||||
INFO("MQTT: receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish\r\n");
|
||||
}
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PINGREQ:
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pingresp(&client->mqtt_state.mqtt_connection);
|
||||
if (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
}
|
||||
break;
|
||||
case MQTT_MSG_TYPE_PINGRESP:
|
||||
// Ignore
|
||||
break;
|
||||
}
|
||||
// NOTE: this is done down here and not in the switch case above
|
||||
// because the PSOCK_READBUF_LEN() won't work inside a switch
|
||||
// statement due to the way protothreads resume.
|
||||
if(msg_type == MQTT_MSG_TYPE_PUBLISH)
|
||||
if (msg_type == MQTT_MSG_TYPE_PUBLISH)
|
||||
{
|
||||
len = client->mqtt_state.message_length_read;
|
||||
len = client->mqtt_state.message_length_read;
|
||||
|
||||
if(client->mqtt_state.message_length < client->mqtt_state.message_length_read)
|
||||
{
|
||||
//client->connState = MQTT_PUBLISH_RECV;
|
||||
//Not Implement yet
|
||||
len -= client->mqtt_state.message_length;
|
||||
pdata += client->mqtt_state.message_length;
|
||||
if (client->mqtt_state.message_length < client->mqtt_state.message_length_read)
|
||||
{
|
||||
//client->connState = MQTT_PUBLISH_RECV;
|
||||
//Not Implement yet
|
||||
len -= client->mqtt_state.message_length;
|
||||
pdata += client->mqtt_state.message_length;
|
||||
|
||||
INFO("Get another published message\r\n");
|
||||
goto READPACKET;
|
||||
}
|
||||
INFO("Get another published message\r\n");
|
||||
goto READPACKET;
|
||||
}
|
||||
|
||||
}
|
||||
break;
|
||||
|
|
@ -257,8 +265,8 @@ mqtt_tcpclient_sent_cb(void *arg)
|
|||
MQTT_Client* client = (MQTT_Client *)pCon->reverse;
|
||||
INFO("TCP: Sent\r\n");
|
||||
client->sendTimeout = 0;
|
||||
if(client->connState == MQTT_DATA && client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH){
|
||||
if(client->publishedCb)
|
||||
if (client->connState == MQTT_DATA && client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH) {
|
||||
if (client->publishedCb)
|
||||
client->publishedCb((uint32_t*)client);
|
||||
}
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
|
|
@ -268,9 +276,9 @@ void ICACHE_FLASH_ATTR mqtt_timer(void *arg)
|
|||
{
|
||||
MQTT_Client* client = (MQTT_Client*)arg;
|
||||
|
||||
if(client->connState == MQTT_DATA){
|
||||
if (client->connState == MQTT_DATA) {
|
||||
client->keepAliveTick ++;
|
||||
if(client->keepAliveTick > client->mqtt_state.connect_info->keepalive){
|
||||
if (client->keepAliveTick > client->mqtt_state.connect_info->keepalive) {
|
||||
|
||||
INFO("\r\nMQTT: Send keepalive packet to %s:%d!\r\n", client->host, client->port);
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pingreq(&client->mqtt_state.mqtt_connection);
|
||||
|
|
@ -280,12 +288,16 @@ void ICACHE_FLASH_ATTR mqtt_timer(void *arg)
|
|||
|
||||
|
||||
client->sendTimeout = MQTT_SEND_TIMOUT;
|
||||
INFO("MQTT: Sending, type: %d, id: %04X\r\n",client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
|
||||
if(client->security){
|
||||
espconn_secure_sent(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
INFO("MQTT: Sending, type: %d, id: %04X\r\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
|
||||
if (client->security) {
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
#else
|
||||
INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else{
|
||||
espconn_sent(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
else {
|
||||
espconn_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
}
|
||||
|
||||
client->mqtt_state.outbound_message = NULL;
|
||||
|
|
@ -294,15 +306,15 @@ void ICACHE_FLASH_ATTR mqtt_timer(void *arg)
|
|||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
}
|
||||
|
||||
} else if(client->connState == TCP_RECONNECT_REQ){
|
||||
} else if (client->connState == TCP_RECONNECT_REQ) {
|
||||
client->reconnectTick ++;
|
||||
if(client->reconnectTick > MQTT_RECONNECT_TIMEOUT) {
|
||||
if (client->reconnectTick > MQTT_RECONNECT_TIMEOUT) {
|
||||
client->reconnectTick = 0;
|
||||
client->connState = TCP_RECONNECT;
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
}
|
||||
}
|
||||
if(client->sendTimeout > 0)
|
||||
if (client->sendTimeout > 0)
|
||||
client->sendTimeout --;
|
||||
}
|
||||
|
||||
|
|
@ -314,7 +326,7 @@ mqtt_tcpclient_discon_cb(void *arg)
|
|||
MQTT_Client* client = (MQTT_Client *)pespconn->reverse;
|
||||
INFO("TCP: Disconnected callback\r\n");
|
||||
client->connState = TCP_RECONNECT_REQ;
|
||||
if(client->disconnectedCb)
|
||||
if (client->disconnectedCb)
|
||||
client->disconnectedCb((uint32_t*)client);
|
||||
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
|
|
@ -345,12 +357,16 @@ mqtt_tcpclient_connect_cb(void *arg)
|
|||
|
||||
|
||||
client->sendTimeout = MQTT_SEND_TIMOUT;
|
||||
INFO("MQTT: Sending, type: %d, id: %04X\r\n",client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
|
||||
if(client->security){
|
||||
espconn_secure_sent(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
INFO("MQTT: Sending, type: %d, id: %04X\r\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
|
||||
if (client->security) {
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
#else
|
||||
INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else{
|
||||
espconn_sent(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
else {
|
||||
espconn_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
|
||||
}
|
||||
|
||||
client->mqtt_state.outbound_message = NULL;
|
||||
|
|
@ -393,17 +409,17 @@ MQTT_Publish(MQTT_Client *client, const char* topic, const char* data, int data_
|
|||
uint8_t dataBuffer[MQTT_BUF_SIZE];
|
||||
uint16_t dataLen;
|
||||
client->mqtt_state.outbound_message = mqtt_msg_publish(&client->mqtt_state.mqtt_connection,
|
||||
topic, data, data_length,
|
||||
qos, retain,
|
||||
&client->mqtt_state.pending_msg_id);
|
||||
if(client->mqtt_state.outbound_message->length == 0){
|
||||
topic, data, data_length,
|
||||
qos, retain,
|
||||
&client->mqtt_state.pending_msg_id);
|
||||
if (client->mqtt_state.outbound_message->length == 0) {
|
||||
INFO("MQTT: Queuing publish failed\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
INFO("MQTT: queuing publish, length: %d, queue size(%d/%d)\r\n", client->mqtt_state.outbound_message->length, client->msgQueue.rb.fill_cnt, client->msgQueue.rb.size);
|
||||
while(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){
|
||||
while (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
if(QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
INFO("MQTT: Serious buffer error\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
|
|
@ -426,9 +442,36 @@ MQTT_Subscribe(MQTT_Client *client, char* topic, uint8_t qos)
|
|||
uint16_t dataLen;
|
||||
|
||||
client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
|
||||
topic, 0,
|
||||
&client->mqtt_state.pending_msg_id);
|
||||
INFO("MQTT: queue subscribe, topic\"%s\", id: %d\r\n",topic, client->mqtt_state.pending_msg_id);
|
||||
topic, 0,
|
||||
&client->mqtt_state.pending_msg_id);
|
||||
INFO("MQTT: queue subscribe, topic\"%s\", id: %d\r\n", topic, client->mqtt_state.pending_msg_id);
|
||||
while (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
INFO("MQTT: Serious buffer error\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
}
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief MQTT ping function.
|
||||
* @param client: MQTT_Client reference
|
||||
* @retval TRUE if success queue
|
||||
*/
|
||||
BOOL ICACHE_FLASH_ATTR
|
||||
MQTT_ping(MQTT_Client *client)
|
||||
{
|
||||
uint8_t dataBuffer[MQTT_BUF_SIZE];
|
||||
uint16_t dataLen;
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pingreq(&client->mqtt_state.mqtt_connection);
|
||||
if(client->mqtt_state.outbound_message->length == 0){
|
||||
INFO("MQTT: Queuing publish failed\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
INFO("MQTT: queuing publish, length: %d, queue size(%d/%d)\r\n", client->mqtt_state.outbound_message->length, client->msgQueue.rb.fill_cnt, client->msgQueue.rb.size);
|
||||
while(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){
|
||||
INFO("MQTT: Queue full\r\n");
|
||||
if(QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
|
|
@ -446,9 +489,9 @@ MQTT_Task(os_event_t *e)
|
|||
MQTT_Client* client = (MQTT_Client*)e->par;
|
||||
uint8_t dataBuffer[MQTT_BUF_SIZE];
|
||||
uint16_t dataLen;
|
||||
if(e->par == 0)
|
||||
if (e->par == 0)
|
||||
return;
|
||||
switch(client->connState){
|
||||
switch (client->connState) {
|
||||
|
||||
case TCP_RECONNECT_REQ:
|
||||
break;
|
||||
|
|
@ -458,21 +501,25 @@ MQTT_Task(os_event_t *e)
|
|||
client->connState = TCP_CONNECTING;
|
||||
break;
|
||||
case MQTT_DATA:
|
||||
if(QUEUE_IsEmpty(&client->msgQueue) || client->sendTimeout != 0) {
|
||||
if (QUEUE_IsEmpty(&client->msgQueue) || client->sendTimeout != 0) {
|
||||
break;
|
||||
}
|
||||
if(QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == 0){
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == 0) {
|
||||
client->mqtt_state.pending_msg_type = mqtt_get_type(dataBuffer);
|
||||
client->mqtt_state.pending_msg_id = mqtt_get_id(dataBuffer, dataLen);
|
||||
|
||||
|
||||
client->sendTimeout = MQTT_SEND_TIMOUT;
|
||||
INFO("MQTT: Sending, type: %d, id: %04X\r\n",client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
|
||||
if(client->security){
|
||||
espconn_secure_sent(client->pCon, dataBuffer, dataLen);
|
||||
INFO("MQTT: Sending, type: %d, id: %04X\r\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
|
||||
if (client->security) {
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_send(client->pCon, dataBuffer, dataLen);
|
||||
#else
|
||||
INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else{
|
||||
espconn_sent(client->pCon, dataBuffer, dataLen);
|
||||
else {
|
||||
espconn_send(client->pCon, dataBuffer, dataLen);
|
||||
}
|
||||
|
||||
client->mqtt_state.outbound_message = NULL;
|
||||
|
|
@ -491,7 +538,7 @@ MQTT_Task(os_event_t *e)
|
|||
* @retval None
|
||||
*/
|
||||
void ICACHE_FLASH_ATTR
|
||||
MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32 port, uint8_t security)
|
||||
MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32_t port, uint8_t security)
|
||||
{
|
||||
uint32_t temp;
|
||||
INFO("MQTT_InitConnection\r\n");
|
||||
|
|
@ -599,12 +646,18 @@ MQTT_Connect(MQTT_Client *mqttClient)
|
|||
os_timer_setfn(&mqttClient->mqttTimer, (os_timer_func_t *)mqtt_timer, mqttClient);
|
||||
os_timer_arm(&mqttClient->mqttTimer, 1000, 1);
|
||||
|
||||
if(UTILS_StrToIP(mqttClient->host, &mqttClient->pCon->proto.tcp->remote_ip)) {
|
||||
if (UTILS_StrToIP(mqttClient->host, &mqttClient->pCon->proto.tcp->remote_ip)) {
|
||||
INFO("TCP: Connect to ip %s:%d\r\n", mqttClient->host, mqttClient->port);
|
||||
if(mqttClient->security){
|
||||
if (mqttClient->security)
|
||||
{
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_connect(mqttClient->pCon);
|
||||
#else
|
||||
INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
else
|
||||
{
|
||||
espconn_connect(mqttClient->pCon);
|
||||
}
|
||||
}
|
||||
|
|
@ -618,9 +671,9 @@ MQTT_Connect(MQTT_Client *mqttClient)
|
|||
void ICACHE_FLASH_ATTR
|
||||
MQTT_Disconnect(MQTT_Client *mqttClient)
|
||||
{
|
||||
if(mqttClient->pCon){
|
||||
if (mqttClient->pCon) {
|
||||
INFO("Free memory\r\n");
|
||||
if(mqttClient->pCon->proto.tcp)
|
||||
if (mqttClient->pCon->proto.tcp)
|
||||
os_free(mqttClient->pCon->proto.tcp);
|
||||
os_free(mqttClient->pCon);
|
||||
mqttClient->pCon = NULL;
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue