mirror
/
esp_mqtt
kopia lustrzana https://github.com/martin-ger/esp_mqtt
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

Merge pull request #92 from vowstar/master 

Adjust disconnect add timeout callback in MQTT module.
develop
Tuan PM 2016-02-02 13:09:12 +07:00
rodzic 70d19cbb4d b71b014b25
commit a5f072bcb1
9 zmienionych plików z 348 dodań i 151 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

9
.travis.yml
Wyświetl plik

@ -1,14 +1,15 @@
language: cpp
before_install:
- sudo apt-get -qq update
- sudo apt-get install -y python-serial srecord
install:
- wget https://github.com/GeorgeHahn/nodemcu-firmware/raw/travis/tools/esp-open-sdk.tar.gz -O tools/esp-open-sdk.tar.gz
- tar -zxvf tools/esp-open-sdk.tar.gz
- export PATH=$PATH:$PWD/esp-open-sdk/sdk:$PWD/esp-open-sdk/xtensa-lx106-elf/bin
- wget http://bbs.espressif.com/download/file.php?id=189 -O tools/esp_iot_sdk_v0.9.5_15_01_23.zip
- unzip tools/esp_iot_sdk_v0.9.5_15_01_23.zip
- wget http://bbs.espressif.com/download/file.php?id=1046 -O tools/esp_iot_sdk_v1.5.1.zip
- unzip tools/esp_iot_sdk_v1.5.1.zip
script:
- make all SDK_BASE="$PWD/esp_iot_sdk_v0.9.5"
- make all SDK_BASE="$PWD/esp_iot_sdk_v1.5.1"
- cd firmware/
- file_name="esp_mqtt_v${TRAVIS_TAG}.${TRAVIS_BUILD_NUMBER}.bin"
- srec_cat -output ${file_name} -binary 0x00000.bin -binary -fill 0xff 0x00000 0x40000 0x40000.bin -binary -offset 0x40000
@ -19,4 +20,4 @@ file: "$TRAVIS_BUILD_DIR/firmware/${file_name}"
skip_cleanup: true
on:
tags: true
repo: tuanpmt/esp_mqtt
repo: tuanpmt/esp_mqtt

2
Makefile
Wyświetl plik

@ -101,7 +101,7 @@ MODULES = driver mqtt user modules
EXTRA_INCDIR = include $(SDK_BASE)/../include
# libraries used in this project, mainly provided by the SDK
LIBS = c gcc hal phy pp net80211 lwip wpa main ssl
LIBS = c gcc hal phy pp net80211 lwip wpa main ssl crypto
# compiler flags using during compilation of source files
CFLAGS = -Os -Wpointer-arith -Wundef -Werror -Wl,-EL -fno-inline-functions -nostdlib -mlongcalls -mtext-section-literals -D__ets__ -DICACHE_FLASH

31
include/mqtt_config.h 100644
Wyświetl plik

@ -0,0 +1,31 @@
#ifndef __MQTT_CONFIG_H__
#define __MQTT_CONFIG_H__
#define CFG_HOLDER 0x00FF55A4 /* Change this value to load default configurations */
#define CFG_LOCATION 0x3C /* Please don't change or if you know what you doing */
#define MQTT_SSL_ENABLE
/*DEFAULT CONFIGURATIONS*/
#define MQTT_HOST "192.168.1.100" //or "mqtt.yourdomain.com"
#define MQTT_PORT 1883
#define MQTT_BUF_SIZE 1024
#define MQTT_KEEPALIVE 120 /*second*/
#define MQTT_CLIENT_ID "DVES_%08X"
#define MQTT_USER "DVES_USER"
#define MQTT_PASS "DVES_PASS"
#define STA_SSID "DVES_HOME"
#define STA_PASS "yourpassword"
#define STA_TYPE AUTH_WPA2_PSK
#define MQTT_RECONNECT_TIMEOUT 5 /*second*/
#define DEFAULT_SECURITY 0
#define QUEUE_BUFFER_SIZE 2048
#define PROTOCOL_NAMEv31 /*MQTT version 3.1 compatible with Mosquitto v0.15*/
//PROTOCOL_NAMEv311 /*MQTT version 3.11 compatible with https://eclipse.org/paho/clients/testing/*/
#endif // __MQTT_CONFIG_H__

37
include/user_config.h
Wyświetl plik

@ -1,30 +1,7 @@
#ifndef _USER_CONFIG_H_
#define _USER_CONFIG_H_
#define CFG_HOLDER 0x00FF55A4 /* Change this value to load default configurations */
#define CFG_LOCATION 0x3C /* Please don't change or if you know what you doing */
#define CLIENT_SSL_ENABLE
/*DEFAULT CONFIGURATIONS*/
#define MQTT_HOST "192.168.11.122" //or "mqtt.yourdomain.com"
#define MQTT_PORT 1880
#define MQTT_BUF_SIZE 1024
#define MQTT_KEEPALIVE 120 /*second*/
#define MQTT_CLIENT_ID "DVES_%08X"
#define MQTT_USER "DVES_USER"
#define MQTT_PASS "DVES_PASS"
#define STA_SSID "DVES_HOME"
#define STA_PASS "yourpassword"
#define STA_TYPE AUTH_WPA2_PSK
#define MQTT_RECONNECT_TIMEOUT 5 /*second*/
#define DEFAULT_SECURITY 0
#define QUEUE_BUFFER_SIZE 2048
#define PROTOCOL_NAMEv31 /*MQTT version 3.1 compatible with Mosquitto v0.15*/
//PROTOCOL_NAMEv311 /*MQTT version 3.11 compatible with https://eclipse.org/paho/clients/testing/*/
#endif
#ifndef __USER_CONFIG_H__
#define __USER_CONFIG_H__
#define USE_OPTIMIZE_PRINTF
#endif

12
mqtt/include/debug.h
Wyświetl plik

@ -8,8 +8,16 @@
#ifndef USER_DEBUG_H_
#define USER_DEBUG_H_
#ifndef INFO
#define INFO os_printf
#if defined(GLOBAL_DEBUG_ON)
#define MQTT_DEBUG_ON
#endif
#if defined(MQTT_DEBUG_ON)
#define INFO( format, ... ) os_printf( format, ## __VA_ARGS__ )
#else
#define INFO( format, ... )
#endif
// #ifndef INFO
// #define INFO os_printf
// #endif
#endif /* USER_DEBUG_H_ */

11
mqtt/include/mqtt.h
Wyświetl plik

@ -29,6 +29,7 @@
*/
#ifndef USER_AT_MQTT_H_
#define USER_AT_MQTT_H_
#include "mqtt_config.h"
#include "mqtt_msg.h"
#include "user_interface.h"
@ -67,6 +68,7 @@ typedef enum {
WIFI_CONNECTING_ERROR,
WIFI_CONNECTED,
DNS_RESOLVE,
TCP_DISCONNECTING,
TCP_DISCONNECTED,
TCP_RECONNECT_REQ,
TCP_RECONNECT,
@ -79,7 +81,9 @@ typedef enum {
MQTT_SUBSCIBE_SENDING,
MQTT_DATA,
MQTT_PUBLISH_RECV,
MQTT_PUBLISHING
MQTT_PUBLISHING,
MQTT_DELETING,
MQTT_DELETED,
} tConnState;
typedef void (*MqttCallback)(uint32_t *args);
@ -96,6 +100,7 @@ typedef struct {
MqttCallback connectedCb;
MqttCallback disconnectedCb;
MqttCallback publishedCb;
MqttCallback timeoutCb;
MqttDataCallback dataCb;
ETSTimer mqttTimer;
uint32_t keepAliveTick;
@ -123,12 +128,14 @@ typedef struct {
#define MQTT_EVENT_TYPE_EXITED 7
#define MQTT_EVENT_TYPE_PUBLISH_CONTINUATION 8
void ICACHE_FLASH_ATTR MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32 port, uint8_t security);
void ICACHE_FLASH_ATTR MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32_t port, uint8_t security);
void ICACHE_FLASH_ATTR MQTT_InitClient(MQTT_Client *mqttClient, uint8_t* client_id, uint8_t* client_user, uint8_t* client_pass, uint32_t keepAliveTime, uint8_t cleanSession);
void ICACHE_FLASH_ATTR MQTT_DeleteClient(MQTT_Client *mqttClient);
void ICACHE_FLASH_ATTR MQTT_InitLWT(MQTT_Client *mqttClient, uint8_t* will_topic, uint8_t* will_msg, uint8_t will_qos, uint8_t will_retain);
void ICACHE_FLASH_ATTR MQTT_OnConnected(MQTT_Client *mqttClient, MqttCallback connectedCb);
void ICACHE_FLASH_ATTR MQTT_OnDisconnected(MQTT_Client *mqttClient, MqttCallback disconnectedCb);
void ICACHE_FLASH_ATTR MQTT_OnPublished(MQTT_Client *mqttClient, MqttCallback publishedCb);
void ICACHE_FLASH_ATTR MQTT_OnTimeout(MQTT_Client *mqttClient, MqttCallback timeoutCb);
void ICACHE_FLASH_ATTR MQTT_OnData(MQTT_Client *mqttClient, MqttDataCallback dataCb);
BOOL ICACHE_FLASH_ATTR MQTT_Subscribe(MQTT_Client *client, char* topic, uint8_t qos);
void ICACHE_FLASH_ATTR MQTT_Connect(MQTT_Client *mqttClient);

1
mqtt/include/mqtt_msg.h
Wyświetl plik

@ -7,6 +7,7 @@
#ifndef MQTT_MSG_H
#define MQTT_MSG_H
#include "mqtt_config.h"
#include "c_types.h"
#ifdef __cplusplus
extern "C" {

394
mqtt/mqtt.c
Wyświetl plik

@ -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,11 +108,84 @@ 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);
}
/**
* @brief Delete tcp client and free all memory
* @param mqttClient: The mqtt client which contain TCP client
* @retval None
*/
void ICACHE_FLASH_ATTR
mqtt_tcpclient_delete(MQTT_Client *mqttClient)
{
if (mqttClient->pCon != NULL) {
INFO("Free memory\r\n");
espconn_delete(mqttClient->pCon);
if (mqttClient->pCon->proto.tcp)
os_free(mqttClient->pCon->proto.tcp);
os_free(mqttClient->pCon);
mqttClient->pCon = NULL;
}
}
/**
* @brief Delete MQTT client and free all memory
* @param mqttClient: The mqtt client
* @retval None
*/
void ICACHE_FLASH_ATTR
mqtt_client_delete(MQTT_Client *mqttClient)
{
mqtt_tcpclient_delete(mqttClient);
if (mqttClient->host != NULL) {
os_free(mqttClient->host);
mqttClient->host = NULL;
}
if (mqttClient->user_data != NULL) {
os_free(mqttClient->user_data);
mqttClient->user_data = NULL;
}
if(mqttClient->connect_info.client_id != NULL) {
os_free(mqttClient->connect_info.client_id);
mqttClient->connect_info.client_id = NULL;
}
if(mqttClient->connect_info.username != NULL) {
os_free(mqttClient->connect_info.username);
mqttClient->connect_info.username = NULL;
}
if(mqttClient->connect_info.password != NULL) {
os_free(mqttClient->connect_info.password);
mqttClient->connect_info.password = NULL;
}
if(mqttClient->connect_info.will_topic != NULL) {
os_free(mqttClient->connect_info.will_topic);
mqttClient->connect_info.will_topic = NULL;
}
if(mqttClient->connect_info.will_message != NULL) {
os_free(mqttClient->connect_info.will_message);
mqttClient->connect_info.will_message = NULL;
}
if(mqttClient->mqtt_state.in_buffer != NULL) {
os_free(mqttClient->mqtt_state.in_buffer);
mqttClient->mqtt_state.in_buffer = NULL;
}
if(mqttClient->mqtt_state.out_buffer != NULL) {
os_free(mqttClient->mqtt_state.out_buffer);
mqttClient->mqtt_state.out_buffer = NULL;
}
}
/**
* @brief Client received callback function.
@ -129,19 +206,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 +230,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 +241,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 +338,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 +349,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,29 +361,41 @@ 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);
err_t result = ESPCONN_OK;
if (client->security) {
#ifdef MQTT_SSL_ENABLE
result = 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 {
result = espconn_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
}
client->mqtt_state.outbound_message = NULL;
client->keepAliveTick = 0;
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
if(ESPCONN_OK == result) {
client->keepAliveTick = 0;
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
}
else {
client->connState = TCP_RECONNECT_REQ;
}
}
} 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->timeoutCb)
client->timeoutCb((uint32_t*)client);
}
}
if(client->sendTimeout > 0)
if (client->sendTimeout > 0)
client->sendTimeout --;
}
@ -313,8 +406,16 @@ mqtt_tcpclient_discon_cb(void *arg)
struct espconn *pespconn = (struct espconn *)arg;
MQTT_Client* client = (MQTT_Client *)pespconn->reverse;
INFO("TCP: Disconnected callback\r\n");
client->connState = TCP_RECONNECT_REQ;
if(client->disconnectedCb)
if(TCP_DISCONNECTING == client->connState) {
client->connState = TCP_DISCONNECTED;
}
else if(MQTT_DELETING == client->connState) {
client->connState = MQTT_DELETED;
}
else {
client->connState = TCP_RECONNECT_REQ;
}
if (client->disconnectedCb)
client->disconnectedCb((uint32_t*)client);
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
@ -345,12 +446,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 +498,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 +531,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, qos,
&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 +578,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;
@ -457,22 +589,47 @@ MQTT_Task(os_event_t *e)
INFO("TCP: Reconnect to: %s:%d\r\n", client->host, client->port);
client->connState = TCP_CONNECTING;
break;
case MQTT_DELETING:
case TCP_DISCONNECTING:
if (client->security) {
#ifdef MQTT_SSL_ENABLE
espconn_secure_connect(client->pCon);
#else
INFO("TCP: Do not support SSL\r\n");
#endif
}
else {
espconn_disconnect(client->pCon);
}
break;
case TCP_DISCONNECTED:
INFO("MQTT: Disconnected\r\n");
mqtt_tcpclient_delete(client);
break;
case MQTT_DELETED:
INFO("MQTT: Deleted client\r\n");
mqtt_client_delete(client);
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 +648,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 +756,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,16 +781,19 @@ MQTT_Connect(MQTT_Client *mqttClient)
void ICACHE_FLASH_ATTR
MQTT_Disconnect(MQTT_Client *mqttClient)
{
if(mqttClient->pCon){
INFO("Free memory\r\n");
if(mqttClient->pCon->proto.tcp)
os_free(mqttClient->pCon->proto.tcp);
os_free(mqttClient->pCon);
mqttClient->pCon = NULL;
}
mqttClient->connState = TCP_DISCONNECTING;
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient);
os_timer_disarm(&mqttClient->mqttTimer);
}
void ICACHE_FLASH_ATTR
MQTT_DeleteClient(MQTT_Client *mqttClient)
{
mqttClient->connState = MQTT_DELETING;
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient);
os_timer_disarm(&mqttClient->mqttTimer);
}
void ICACHE_FLASH_ATTR
MQTT_OnConnected(MQTT_Client *mqttClient, MqttCallback connectedCb)
{
@ -651,3 +817,9 @@ MQTT_OnPublished(MQTT_Client *mqttClient, MqttCallback publishedCb)
{
mqttClient->publishedCb = publishedCb;
}
void ICACHE_FLASH_ATTR
MQTT_OnTimeout(MQTT_Client *mqttClient, MqttCallback timeoutCb)
{
mqttClient->timeoutCb = timeoutCb;
}

2
mqtt/mqtt_msg.c
Wyświetl plik

@ -127,7 +127,7 @@ static mqtt_message_t* ICACHE_FLASH_ATTR fini_message(mqtt_connection_t* connect
void ICACHE_FLASH_ATTR mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length)
{
memset(connection, 0, sizeof(connection));
memset(connection, 0, sizeof(mqtt_connection_t));
connection->buffer = buffer;
connection->buffer_length = buffer_length;
}