mqtt.c File Reference

#include <esp8266.h>
#include <lwip/ip.h>
#include <lwip/udp.h>
#include <lwip/tcp_impl.h>
#include <netif/etharp.h>
#include <lwip/netif.h>
#include <lwip/lwip_napt.h>
#include <lwip/dns.h>
#include <lwip/app/dhcpserver.h>
#include <lwip/opt.h>
#include <espconn.h>
#include "mqtt_msg.h"
#include "debug.h"
#include "user_config.h"
#include "mqtt.h"
#include "mqtt_utils.h"
#include "queue.h"

Include dependency graph for mqtt.c:

Go to the source code of this file.

Macros
#define	MQTT_TASK_PRIO   USER_TASK_PRIO_1
#define	MQTT_TASK_QUEUE_SIZE   100
#define	MQTT_SEND_TIMOUT   5
#define	QUEUE_BUFFER_SIZE   2048

Functions
LOCAL void ICACHE_FLASH_ATTR	mqtt_dns_found (const char *name, ip_addr_t *ipaddr, void *arg)
LOCAL void ICACHE_FLASH_ATTR	deliver_publish (MQTT_Client *client, uint8_t *message, int length)
void ICACHE_FLASH_ATTR	mqtt_send_keepalive (MQTT_Client *client)
void ICACHE_FLASH_ATTR	mqtt_tcpclient_delete (MQTT_Client *mqttClient)
	Delete tcp client and free all memory. More...
void ICACHE_FLASH_ATTR	mqtt_client_delete (MQTT_Client *mqttClient)
	Delete MQTT client and free all memory. More...
void ICACHE_FLASH_ATTR	mqtt_tcpclient_recv (void *arg, char *pdata, unsigned short len)
	Client received callback function. More...
void ICACHE_FLASH_ATTR	mqtt_tcpclient_sent_cb (void *arg)
	Client send over callback function. More...
void ICACHE_FLASH_ATTR	mqtt_timer (void *arg)
void ICACHE_FLASH_ATTR	mqtt_tcpclient_discon_cb (void *arg)
void ICACHE_FLASH_ATTR	mqtt_tcpclient_connect_cb (void *arg)
	Tcp client connect success callback function. More...
void ICACHE_FLASH_ATTR	mqtt_tcpclient_recon_cb (void *arg, sint8 errType)
	Tcp client connect repeat callback function. More...
BOOL ICACHE_FLASH_ATTR	MQTT_Publish (MQTT_Client *client, const char *topic, const char *data, int data_length, int qos, int retain)
	MQTT publish function. More...
BOOL ICACHE_FLASH_ATTR	MQTT_Subscribe (MQTT_Client *client, char *topic, uint8_t qos)
	MQTT subscibe function. More...
BOOL ICACHE_FLASH_ATTR	MQTT_UnSubscribe (MQTT_Client *client, char *topic)
	MQTT un-subscibe function. More...
BOOL ICACHE_FLASH_ATTR	MQTT_Ping (MQTT_Client *client)
	MQTT ping function. More...
void ICACHE_FLASH_ATTR	MQTT_Task (os_event_t *e)
void ICACHE_FLASH_ATTR	MQTT_InitConnection (MQTT_Client *mqttClient, uint8_t *host, uint32_t port, uint8_t security)
	MQTT initialization connection function. More...
BOOL 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)
	MQTT initialization mqtt client function. More...
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_Connect (MQTT_Client *mqttClient)
	Begin connect to MQTT broker. More...
void ICACHE_FLASH_ATTR	MQTT_Disconnect (MQTT_Client *mqttClient)
void ICACHE_FLASH_ATTR	MQTT_DeleteClient (MQTT_Client *mqttClient)
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_OnData (MQTT_Client *mqttClient, MqttDataCallback dataCb)
void ICACHE_FLASH_ATTR	MQTT_OnPublished (MQTT_Client *mqttClient, MqttCallback publishedCb)
void ICACHE_FLASH_ATTR	MQTT_OnTimeout (MQTT_Client *mqttClient, MqttCallback timeoutCb)

Variables
unsigned char *	default_certificate
unsigned int	default_certificate_len = 0
unsigned char *	default_private_key
unsigned int	default_private_key_len = 0
os_event_t	mqtt_procTaskQueue [MQTT_TASK_QUEUE_SIZE]

Macro Definition Documentation

MQTT_SEND_TIMOUT

#define MQTT_SEND_TIMOUT   5

Definition at line 58 of file mqtt.c.

Referenced by mqtt_send_keepalive(), MQTT_Task(), and mqtt_tcpclient_connect_cb().

MQTT_TASK_PRIO

#define MQTT_TASK_PRIO   USER_TASK_PRIO_1

Definition at line 56 of file mqtt.c.

Referenced by MQTT_DeleteClient(), MQTT_Disconnect(), mqtt_dns_found(), MQTT_InitClient(), MQTT_Ping(), MQTT_Publish(), mqtt_send_keepalive(), MQTT_Subscribe(), mqtt_tcpclient_connect_cb(), mqtt_tcpclient_discon_cb(), mqtt_tcpclient_recon_cb(), mqtt_tcpclient_recv(), mqtt_tcpclient_sent_cb(), mqtt_timer(), and MQTT_UnSubscribe().

MQTT_TASK_QUEUE_SIZE

#define MQTT_TASK_QUEUE_SIZE   100

Definition at line 57 of file mqtt.c.

Referenced by MQTT_InitClient().

QUEUE_BUFFER_SIZE

#define QUEUE_BUFFER_SIZE   2048

Definition at line 61 of file mqtt.c.

Referenced by MQTT_InitClient().

Function Documentation

deliver_publish()

LOCAL void ICACHE_FLASH_ATTR deliver_publish	(	MQTT_Client *	client,
		uint8_t *	message,
		int	length
	)

Definition at line 119 of file mqtt.c.

References mqtt_event_data_t::data, mqtt_event_data_t::data_length, MQTT_Client::dataCb, ICACHE_FLASH_ATTR, mqtt_get_publish_data(), mqtt_get_publish_topic(), mqtt_event_data_t::topic, and mqtt_event_data_t::topic_length.

Referenced by mqtt_tcpclient_recv().

{
    mqtt_event_data_t event_data;

    event_data.topic_length = length;
    event_data.topic = mqtt_get_publish_topic(message, &event_data.topic_length);
    event_data.data_length = length;
    event_data.data = mqtt_get_publish_data(message, &event_data.data_length);

    if (client->dataCb)
        client->dataCb((uint32_t*)client, event_data.topic, event_data.topic_length, event_data.data, event_data.data_length);

}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_client_delete()

void ICACHE_FLASH_ATTR mqtt_client_delete

(

MQTT_Client *

mqttClient

)

Delete MQTT client and free all memory.

Parameters

mqttClient

The mqtt client

Return values

None

Definition at line 203 of file mqtt.c.

References QUEUE::buf, mqtt_connection::buffer, mqtt_connect_info::client_id, MQTT_Client::connect_info, MQTT_Client::connectedCb, MQTT_Client::connState, mqtt_message::data, MQTT_Client::dataCb, MQTT_Client::disconnectedCb, MQTT_Client::host, ICACHE_FLASH_ATTR, mqtt_state_t::in_buffer, INFO, mqtt_state_t::mqtt_connection, MQTT_Client::mqtt_state, mqtt_tcpclient_delete(), MQTT_Client::msgQueue, NULL, os_free, mqtt_state_t::out_buffer, mqtt_state_t::outbound_message, mqtt_connect_info::password, MQTT_Client::pCon, MQTT_Client::publishedCb, MQTT_Client::timeoutCb, MQTT_Client::user_data, mqtt_connect_info::username, WIFI_INIT, mqtt_connect_info::will_message, and mqtt_connect_info::will_topic.

Referenced by MQTT_Task().

{
    if(mqttClient == NULL)
        return;

    if (mqttClient->pCon != NULL){
        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->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;
    }

    if(mqttClient->mqtt_state.outbound_message != NULL) {
        if(mqttClient->mqtt_state.outbound_message->data != NULL)
        {
            os_free(mqttClient->mqtt_state.outbound_message->data);
            mqttClient->mqtt_state.outbound_message->data = NULL;
        }
    }

    if(mqttClient->mqtt_state.mqtt_connection.buffer != NULL) {
        // Already freed but not NULL
        mqttClient->mqtt_state.mqtt_connection.buffer = 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->msgQueue.buf != NULL) {
        os_free(mqttClient->msgQueue.buf);
        mqttClient->msgQueue.buf = NULL;
    }

    // Initialize state
    mqttClient->connState = WIFI_INIT;
    // Clear callback functions to avoid abnormal callback
    mqttClient->connectedCb = NULL;
    mqttClient->disconnectedCb = NULL;
    mqttClient->publishedCb = NULL;
    mqttClient->timeoutCb = NULL;
    mqttClient->dataCb = NULL;

    INFO("MQTT: client already deleted\r\n");
}

Here is the call graph for this function:

Here is the caller graph for this function:

MQTT_Connect()

void ICACHE_FLASH_ATTR MQTT_Connect

(

MQTT_Client *

mqttClient

)

Begin connect to MQTT broker.

Parameters

client

MQTT_Client reference

Return values

None

Definition at line 892 of file mqtt.c.

References MQTT_Client::connState, espconn_connect(), espconn_gethostbyname(), ESPCONN_KEEPALIVE, ESPCONN_KEEPCNT, ESPCONN_KEEPIDLE, ESPCONN_KEEPINTVL, ESPCONN_NONE, espconn_port(), espconn_regist_connectcb(), espconn_regist_reconcb(), espconn_secure_connect(), espconn_set_keepalive(), espconn_set_opt(), ESPCONN_TCP, MQTT_Client::host, ICACHE_FLASH_ATTR, INFO, MQTT_Client::ip, MQTT_Client::keepAliveTick, _esp_tcp::local_port, mqtt_dns_found(), mqtt_tcpclient_connect_cb(), mqtt_tcpclient_delete(), mqtt_tcpclient_recon_cb(), mqtt_timer(), MQTT_Client::mqttTimer, os_timer_arm, os_timer_disarm, os_timer_func_t, os_timer_setfn, os_zalloc, MQTT_Client::pCon, MQTT_Client::port, espconn::proto, MQTT_Client::reconnectTick, _esp_tcp::remote_ip, _esp_tcp::remote_port, espconn::reverse, MQTT_Client::security, espconn::state, espconn::tcp, TCP_CONNECTING, espconn::type, and UTILS_StrToIP().

Referenced by MQTT_Task().

{
    uint32_t keeplive; 
    
    if (mqttClient->pCon) {
        // Clean up the old connection forcefully - using MQTT_Disconnect
        // does not actually release the old connection until the
        // disconnection callback is invoked.
        mqtt_tcpclient_delete(mqttClient);
    }
    mqttClient->pCon = (struct espconn *)os_zalloc(sizeof(struct espconn));
    mqttClient->pCon->type = ESPCONN_TCP;
    mqttClient->pCon->state = ESPCONN_NONE;
    mqttClient->pCon->proto.tcp = (esp_tcp *)os_zalloc(sizeof(esp_tcp));
    mqttClient->pCon->proto.tcp->local_port = espconn_port();
    mqttClient->pCon->proto.tcp->remote_port = mqttClient->port;
    mqttClient->pCon->reverse = mqttClient;
    
    // enable TCP keep alive
    espconn_set_opt(mqttClient->pCon, ESPCONN_KEEPALIVE);
    
    // set keepalive: 75s = 60 + 5*3
    keeplive = 60;
    espconn_set_keepalive(mqttClient->pCon, ESPCONN_KEEPIDLE, &keeplive);

    keeplive = 5;
    espconn_set_keepalive(mqttClient->pCon, ESPCONN_KEEPINTVL, &keeplive);
    
    keeplive = 3;   //try times
    espconn_set_keepalive(mqttClient->pCon, ESPCONN_KEEPCNT, &keeplive);
    
    espconn_regist_connectcb(mqttClient->pCon, mqtt_tcpclient_connect_cb);
    espconn_regist_reconcb(mqttClient->pCon, mqtt_tcpclient_recon_cb);

    mqttClient->keepAliveTick = 0;
    mqttClient->reconnectTick = 0;

    os_timer_disarm(&mqttClient->mqttTimer);
    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)) {
        INFO("TCP: Connect to ip  %s:%d\r\n", mqttClient->host, mqttClient->port);
        if (mqttClient->security)
        {
#ifdef MQTT_SSL_ENABLE
            espconn_secure_connect(mqttClient->pCon);
#else
            INFO("TCP: Do not support SSL\r\n");
#endif
        }
        else
        {
            espconn_connect(mqttClient->pCon);
        }
    }
    else {
        INFO("TCP: Connect to domain %s:%d\r\n", mqttClient->host, mqttClient->port);
        espconn_gethostbyname(mqttClient->pCon, mqttClient->host, &mqttClient->ip, mqtt_dns_found);
    }
    mqttClient->connState = TCP_CONNECTING;
}

Here is the call graph for this function:

Here is the caller graph for this function:

MQTT_DeleteClient()

void ICACHE_FLASH_ATTR MQTT_DeleteClient

(

MQTT_Client *

mqttClient

)

Definition at line 964 of file mqtt.c.

References MQTT_Client::connState, ICACHE_FLASH_ATTR, MQTT_DELETED, MQTT_TASK_PRIO, NULL, os_param_t, os_timer_disarm, and system_os_post().

{
    if(NULL == mqttClient)
        return;

    mqttClient->connState = MQTT_DELETED;
    // if(TCP_DISCONNECTED == mqttClient->connState) {
    //  mqttClient->connState = MQTT_DELETED;
    // } else if(MQTT_DELETED != mqttClient->connState) {
    //  mqttClient->connState = MQTT_DELETING;  
    // }

    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient);
    os_timer_disarm(&mqttClient->mqttTimer);
}

Here is the call graph for this function:

MQTT_Disconnect()

void ICACHE_FLASH_ATTR MQTT_Disconnect

(

MQTT_Client *

mqttClient

)

Definition at line 956 of file mqtt.c.

References MQTT_Client::connState, ICACHE_FLASH_ATTR, MQTT_TASK_PRIO, os_param_t, os_timer_disarm, system_os_post(), and TCP_DISCONNECTING.

{
    mqttClient->connState = TCP_DISCONNECTING;
    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient);
    os_timer_disarm(&mqttClient->mqttTimer);
}

Here is the call graph for this function:

mqtt_dns_found()

LOCAL void ICACHE_FLASH_ATTR mqtt_dns_found	(	const char *	name,
		ip_addr_t *	ipaddr,
		void *	arg
	)

Definition at line 76 of file mqtt.c.

References MQTT_Client::connState, espconn_connect(), espconn_secure_connect(), ICACHE_FLASH_ATTR, INFO, MQTT_Client::ip, LOCAL, MQTT_TASK_PRIO, NULL, os_memcpy, os_param_t, MQTT_Client::pCon, espconn::proto, _esp_tcp::remote_ip, espconn::reverse, MQTT_Client::security, system_os_post(), espconn::tcp, TCP_CONNECTING, and TCP_RECONNECT_REQ.

Referenced by MQTT_Connect().

{
    struct espconn *pConn = (struct espconn *)arg;
    MQTT_Client* client = (MQTT_Client *)pConn->reverse;


    if (ipaddr == NULL)
    {
        INFO("DNS: Found, but got no ip, try to reconnect\r\n");
        client->connState = TCP_RECONNECT_REQ;
        return;
    }

    INFO("DNS: found ip %d.%d.%d.%d\n",
         *((uint8 *) &ipaddr->addr),
         *((uint8 *) &ipaddr->addr + 1),
         *((uint8 *) &ipaddr->addr + 2),
         *((uint8 *) &ipaddr->addr + 3));

    if (client->ip.addr == 0 && ipaddr->addr != 0)
    {
        os_memcpy(client->pCon->proto.tcp->remote_ip, &ipaddr->addr, 4);
        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);
        }

        client->connState = TCP_CONNECTING;
        INFO("TCP: connecting...\r\n");
    }

    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
}

Here is the call graph for this function:

Here is the caller graph for this function:

MQTT_InitClient()

BOOL 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
	)

MQTT initialization mqtt client function.

Parameters

client	MQTT_Client reference
clientid	MQTT client id
client_user:MQTT	client user
client_pass:MQTT	client password
client_pass:MQTT	keep alive timer, in second

Return values

None

Definition at line 801 of file mqtt.c.

References mqtt_connect_info::clean_session, mqtt_connect_info::client_id, mqtt_state_t::connect_info, MQTT_Client::connect_info, ICACHE_FLASH_ATTR, mqtt_state_t::in_buffer, mqtt_state_t::in_buffer_length, INFO, mqtt_connect_info::keepalive, MQTT_BUF_SIZE, mqtt_state_t::mqtt_connection, mqtt_msg_init(), mqtt_procTaskQueue, MQTT_Client::mqtt_state, MQTT_Task(), MQTT_TASK_PRIO, MQTT_TASK_QUEUE_SIZE, MQTT_Client::msgQueue, os_memset, os_param_t, os_strcpy, os_strlen, os_zalloc, mqtt_state_t::out_buffer, mqtt_state_t::out_buffer_length, mqtt_connect_info::password, QUEUE_BUFFER_SIZE, QUEUE_Init(), system_os_post(), system_os_task(), and mqtt_connect_info::username.

{
    uint32_t temp;
    INFO("MQTT:InitClient\r\n");

    os_memset(&mqttClient->connect_info, 0, sizeof(mqtt_connect_info_t));

    if ( !client_id ) {
        /* Should be allowed by broker, but clean session flag must be set. */
#ifdef PROTOCOL_NAMEv311
        if (cleanSession) {
            mqttClient->connect_info.client_id = zero_len_id;
        } else {
            INFO("cleanSession must be set to use 0 length client_id\r\n");
            return false;
        }
        /* Not supported. Return. */
#else
        INFO("Client ID required for MQTT < 3.1.1!\r\n");
        return false;
#endif
    }

    /* If connect_info's client_id is still NULL and we get here, we can        *
     * assume the passed client_id is non-NULL.                                 */
    if ( !(mqttClient->connect_info.client_id) )
    {
      temp = os_strlen(client_id);
      mqttClient->connect_info.client_id = (uint8_t*)os_zalloc(temp + 1);
      os_strcpy(mqttClient->connect_info.client_id, client_id);
      mqttClient->connect_info.client_id[temp] = 0;
    }

    if (client_user)
    {
        temp = os_strlen(client_user);
        mqttClient->connect_info.username = (uint8_t*)os_zalloc(temp + 1);
        os_strcpy(mqttClient->connect_info.username, client_user);
        mqttClient->connect_info.username[temp] = 0;
    }

    if (client_pass)
    {
        temp = os_strlen(client_pass);
        mqttClient->connect_info.password = (uint8_t*)os_zalloc(temp + 1);
        os_strcpy(mqttClient->connect_info.password, client_pass);
        mqttClient->connect_info.password[temp] = 0;
    }


    mqttClient->connect_info.keepalive = keepAliveTime;
    mqttClient->connect_info.clean_session = cleanSession;

    mqttClient->mqtt_state.in_buffer = (uint8_t *)os_zalloc(MQTT_BUF_SIZE);
    mqttClient->mqtt_state.in_buffer_length = MQTT_BUF_SIZE;
    mqttClient->mqtt_state.out_buffer =  (uint8_t *)os_zalloc(MQTT_BUF_SIZE);
    mqttClient->mqtt_state.out_buffer_length = MQTT_BUF_SIZE;
    mqttClient->mqtt_state.connect_info = &mqttClient->connect_info;

    mqtt_msg_init(&mqttClient->mqtt_state.mqtt_connection, mqttClient->mqtt_state.out_buffer, mqttClient->mqtt_state.out_buffer_length);

    QUEUE_Init(&mqttClient->msgQueue, QUEUE_BUFFER_SIZE);

    system_os_task(MQTT_Task, MQTT_TASK_PRIO, mqtt_procTaskQueue, MQTT_TASK_QUEUE_SIZE);
    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient);
    return true;
}

Here is the call graph for this function:

MQTT_InitConnection()

void ICACHE_FLASH_ATTR MQTT_InitConnection	(	MQTT_Client *	mqttClient,
		uint8_t *	host,
		uint32_t	port,
		uint8_t	security
	)

MQTT initialization connection function.

Parameters

client	MQTT_Client reference
host	Domain or IP string
port	Port to connect
security	1 for ssl, 0 for none

Return values

None

Definition at line 777 of file mqtt.c.

References BOOL, MQTT_Client::host, ICACHE_FLASH_ATTR, INFO, os_memset, os_strcpy, os_strlen, os_zalloc, MQTT_Client::port, and MQTT_Client::security.

{
    uint32_t temp;
    INFO("MQTT:InitConnection\r\n");
    os_memset(mqttClient, 0, sizeof(MQTT_Client));
    temp = os_strlen(host);
    mqttClient->host = (uint8_t*)os_zalloc(temp + 1);
    os_strcpy(mqttClient->host, host);
    mqttClient->host[temp] = 0;
    mqttClient->port = port;
    mqttClient->security = security;

}

MQTT_InitLWT()

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
	)

Definition at line 869 of file mqtt.c.

References MQTT_Client::connect_info, ICACHE_FLASH_ATTR, os_strcpy, os_strlen, os_zalloc, mqtt_connect_info::will_message, mqtt_connect_info::will_qos, mqtt_connect_info::will_retain, and mqtt_connect_info::will_topic.

{
    uint32_t temp;
    temp = os_strlen(will_topic);
    mqttClient->connect_info.will_topic = (uint8_t*)os_zalloc(temp + 1);
    os_strcpy(mqttClient->connect_info.will_topic, will_topic);
    mqttClient->connect_info.will_topic[temp] = 0;

    temp = os_strlen(will_msg);
    mqttClient->connect_info.will_message = (uint8_t*)os_zalloc(temp + 1);
    os_strcpy(mqttClient->connect_info.will_message, will_msg);
    mqttClient->connect_info.will_message[temp] = 0;


    mqttClient->connect_info.will_qos = will_qos;
    mqttClient->connect_info.will_retain = will_retain;
}

MQTT_OnConnected()

void ICACHE_FLASH_ATTR MQTT_OnConnected	(	MQTT_Client *	mqttClient,
		MqttCallback	connectedCb
	)

Definition at line 981 of file mqtt.c.

References MQTT_Client::connectedCb, and ICACHE_FLASH_ATTR.

{
    mqttClient->connectedCb = connectedCb;
}

MQTT_OnData()

void ICACHE_FLASH_ATTR MQTT_OnData	(	MQTT_Client *	mqttClient,
		MqttDataCallback	dataCb
	)

Definition at line 993 of file mqtt.c.

References MQTT_Client::dataCb, and ICACHE_FLASH_ATTR.

{
    mqttClient->dataCb = dataCb;
}

MQTT_OnDisconnected()

void ICACHE_FLASH_ATTR MQTT_OnDisconnected	(	MQTT_Client *	mqttClient,
		MqttCallback	disconnectedCb
	)

Definition at line 987 of file mqtt.c.

References MQTT_Client::disconnectedCb, and ICACHE_FLASH_ATTR.

{
    mqttClient->disconnectedCb = disconnectedCb;
}

MQTT_OnPublished()

void ICACHE_FLASH_ATTR MQTT_OnPublished	(	MQTT_Client *	mqttClient,
		MqttCallback	publishedCb
	)

Definition at line 999 of file mqtt.c.

References ICACHE_FLASH_ATTR, and MQTT_Client::publishedCb.

{
    mqttClient->publishedCb = publishedCb;
}

MQTT_OnTimeout()

void ICACHE_FLASH_ATTR MQTT_OnTimeout	(	MQTT_Client *	mqttClient,
		MqttCallback	timeoutCb
	)

Definition at line 1005 of file mqtt.c.

References MQTT_Client::timeoutCb.

{
    mqttClient->timeoutCb = timeoutCb;
}

MQTT_Ping()

BOOL ICACHE_FLASH_ATTR MQTT_Ping

(

MQTT_Client *

client

)

MQTT ping function.

Parameters

client

MQTT_Client reference

Return values

TRUE	if success queue

Definition at line 675 of file mqtt.c.

References mqtt_message::data, FALSE, RINGBUF::fill_cnt, ICACHE_FLASH_ATTR, INFO, mqtt_message::length, MQTT_BUF_SIZE, mqtt_state_t::mqtt_connection, mqtt_msg_pingreq(), MQTT_Client::mqtt_state, MQTT_TASK_PRIO, MQTT_Client::msgQueue, os_param_t, mqtt_state_t::outbound_message, QUEUE_Gets(), QUEUE_Puts(), QUEUE::rb, RINGBUF::size, system_os_post(), and TRUE.

{
    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(%ld/%ld)\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) {
            INFO("MQTT: Serious buffer error\r\n");
            return FALSE;
        }
    }
    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
    return TRUE;
}

Here is the call graph for this function:

MQTT_Publish()

BOOL ICACHE_FLASH_ATTR MQTT_Publish	(	MQTT_Client *	client,
		const char *	topic,
		const char *	data,
		int	data_length,
		int	qos,
		int	retain
	)

MQTT publish function.

Parameters

client	MQTT_Client reference
topic	string topic will publish to
data	buffer data send point to
data_length	length of data
qos	qos
retain	retain

Return values

TRUE	if success queue

Definition at line 591 of file mqtt.c.

References BOOL, mqtt_message::data, FALSE, RINGBUF::fill_cnt, ICACHE_FLASH_ATTR, INFO, mqtt_message::length, MQTT_BUF_SIZE, mqtt_state_t::mqtt_connection, mqtt_msg_publish(), MQTT_Client::mqtt_state, MQTT_TASK_PRIO, MQTT_Client::msgQueue, os_param_t, mqtt_state_t::outbound_message, mqtt_state_t::pending_msg_id, QUEUE_Gets(), QUEUE_Puts(), QUEUE::rb, RINGBUF::size, system_os_post(), and TRUE.

Referenced by en61107_received_task(), en61107_request_send(), kmp_received_task(), and kmp_request_send().

{
    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) {
        INFO("MQTT: Queuing publish failed\r\n");
        return FALSE;
    }
    INFO("MQTT: queuing publish, length: %d, queue size(%ld/%ld)\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) {
            INFO("MQTT: Serious buffer error\r\n");
            return FALSE;
        }
    }
    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
    return TRUE;
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_send_keepalive()

void ICACHE_FLASH_ATTR mqtt_send_keepalive

(

MQTT_Client *

client

)

Definition at line 134 of file mqtt.c.

References MQTT_Client::connState, mqtt_message::data, ESPCONN_OK, espconn_secure_send(), espconn_send(), ESPCONN_WRITE, MQTT_Client::host, ICACHE_FLASH_ATTR, INFO, MQTT_Client::keepAliveTick, mqtt_message::length, mqtt_state_t::mqtt_connection, MQTT_DATA, mqtt_get_id(), mqtt_get_type(), mqtt_msg_pingreq(), MQTT_MSG_TYPE_PINGREQ, MQTT_SEND_TIMOUT, MQTT_Client::mqtt_state, MQTT_TASK_PRIO, NULL, os_param_t, mqtt_state_t::outbound_message, MQTT_Client::pCon, mqtt_state_t::pending_msg_id, mqtt_state_t::pending_msg_type, MQTT_Client::port, MQTT_Client::security, MQTT_Client::sendTimeout, espconn::state, system_os_post(), and TCP_RECONNECT_DISCONNECTING.

Referenced by MQTT_Task().

                                         {
    if (client->pCon->state == ESPCONN_WRITE) {
        system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
        return;
    }

    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);
    client->mqtt_state.pending_msg_type = MQTT_MSG_TYPE_PINGREQ;
    client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
    client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);


    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);
    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 {
        result = espconn_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
    }

    client->mqtt_state.outbound_message = NULL;
    if(ESPCONN_OK == result) {
        client->keepAliveTick = 0;
        client->connState = MQTT_DATA;
        system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
    }
    else {
        client->connState = TCP_RECONNECT_DISCONNECTING;
        system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

MQTT_Subscribe()

BOOL ICACHE_FLASH_ATTR MQTT_Subscribe	(	MQTT_Client *	client,
		char *	topic,
		uint8_t	qos
	)

MQTT subscibe function.

Parameters

client	MQTT_Client reference
topic	string topic will subscribe
qos	qos

Return values

TRUE	if success queue

Definition at line 623 of file mqtt.c.

References BOOL, mqtt_message::data, FALSE, ICACHE_FLASH_ATTR, INFO, mqtt_message::length, MQTT_BUF_SIZE, mqtt_state_t::mqtt_connection, mqtt_msg_subscribe(), MQTT_Client::mqtt_state, MQTT_TASK_PRIO, MQTT_Client::msgQueue, os_param_t, mqtt_state_t::outbound_message, mqtt_state_t::pending_msg_id, QUEUE_Gets(), QUEUE_Puts(), system_os_post(), and TRUE.

{
    uint8_t dataBuffer[MQTT_BUF_SIZE];
    uint16_t dataLen;

    client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,
                                          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;
}

Here is the call graph for this function:

MQTT_Task()

void ICACHE_FLASH_ATTR MQTT_Task

(

os_event_t *

e

)

Definition at line 697 of file mqtt.c.

References MQTT_Client::connState, espconn_disconnect(), espconn_secure_disconnect(), espconn_secure_send(), espconn_send(), MQTT_Client::host, ICACHE_FLASH_ATTR, INFO, MQTT_BUF_SIZE, mqtt_client_delete(), MQTT_Connect(), MQTT_DATA, MQTT_DELETED, MQTT_DELETING, mqtt_get_id(), mqtt_get_type(), MQTT_KEEPALIVE_SEND, mqtt_send_keepalive(), MQTT_SEND_TIMOUT, MQTT_Client::mqtt_state, mqtt_tcpclient_delete(), MQTT_Client::msgQueue, NULL, mqtt_state_t::outbound_message, MQTT_Client::pCon, mqtt_state_t::pending_msg_id, mqtt_state_t::pending_msg_type, MQTT_Client::port, QUEUE_Gets(), QUEUE_IsEmpty(), MQTT_Client::security, MQTT_Client::sendTimeout, TCP_CONNECTING, TCP_DISCONNECTED, TCP_DISCONNECTING, TCP_RECONNECT, TCP_RECONNECT_DISCONNECTING, and TCP_RECONNECT_REQ.

Referenced by MQTT_InitClient().

{
    MQTT_Client* client = (MQTT_Client*)e->par;
    uint8_t dataBuffer[MQTT_BUF_SIZE];
    uint16_t dataLen;
    if (e->par == 0)
        return;
    switch (client->connState) {

    case TCP_RECONNECT_REQ:
        break;
    case TCP_RECONNECT:
        mqtt_tcpclient_delete(client);
        MQTT_Connect(client);
        INFO("TCP: Reconnect to: %s:%d\r\n", client->host, client->port);
        client->connState = TCP_CONNECTING;
        break;
    case MQTT_DELETING:
    case TCP_DISCONNECTING:
    case TCP_RECONNECT_DISCONNECTING:
        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);
        }
        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_KEEPALIVE_SEND:
        mqtt_send_keepalive(client);
        break;
    case MQTT_DATA:
        if (QUEUE_IsEmpty(&client->msgQueue) || client->sendTimeout != 0) {
            break;
        }
        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) {
#ifdef MQTT_SSL_ENABLE
                espconn_secure_send(client->pCon, dataBuffer, dataLen);
#else
                INFO("TCP: Do not support SSL\r\n");
#endif
            }
            else {
                espconn_send(client->pCon, dataBuffer, dataLen);
            }

            client->mqtt_state.outbound_message = NULL;
            break;
        }
        break;
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_tcpclient_connect_cb()

void ICACHE_FLASH_ATTR mqtt_tcpclient_connect_cb

(

void *

arg

)

Tcp client connect success callback function.

Parameters

arg	contain the ip link information

Return values

None

Definition at line 527 of file mqtt.c.

References mqtt_state_t::connect_info, MQTT_Client::connState, mqtt_message::data, espconn_regist_disconcb(), espconn_regist_recvcb(), espconn_regist_sentcb(), espconn_secure_send(), espconn_send(), MQTT_Client::host, ICACHE_FLASH_ATTR, INFO, mqtt_message::length, MQTT_CONNECT_SENDING, mqtt_state_t::mqtt_connection, mqtt_get_id(), mqtt_get_type(), mqtt_msg_connect(), mqtt_msg_init(), MQTT_SEND_TIMOUT, MQTT_Client::mqtt_state, MQTT_TASK_PRIO, mqtt_tcpclient_discon_cb(), mqtt_tcpclient_recv(), mqtt_tcpclient_sent_cb(), NULL, os_param_t, mqtt_state_t::out_buffer, mqtt_state_t::out_buffer_length, mqtt_state_t::outbound_message, MQTT_Client::pCon, mqtt_state_t::pending_msg_id, mqtt_state_t::pending_msg_type, MQTT_Client::port, espconn::reverse, MQTT_Client::security, MQTT_Client::sendTimeout, and system_os_post().

Referenced by MQTT_Connect().

{
    struct espconn *pCon = (struct espconn *)arg;
    MQTT_Client* client = (MQTT_Client *)pCon->reverse;

    espconn_regist_disconcb(client->pCon, mqtt_tcpclient_discon_cb);
    espconn_regist_recvcb(client->pCon, mqtt_tcpclient_recv);////////
    espconn_regist_sentcb(client->pCon, mqtt_tcpclient_sent_cb);///////
    INFO("MQTT: Connected to broker %s:%d\r\n", client->host, client->port);

    mqtt_msg_init(&client->mqtt_state.mqtt_connection, client->mqtt_state.out_buffer, client->mqtt_state.out_buffer_length);
    client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection, client->mqtt_state.connect_info);
    client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
    client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);


    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) {
#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_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
    }

    client->mqtt_state.outbound_message = NULL;
    client->connState = MQTT_CONNECT_SENDING;
    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_tcpclient_delete()

void ICACHE_FLASH_ATTR mqtt_tcpclient_delete

(

MQTT_Client *

mqttClient

)

Delete tcp client and free all memory.

Parameters

mqttClient

The mqtt client which contain TCP client

Return values

None

Definition at line 179 of file mqtt.c.

References espconn_abort(), espconn_delete(), ICACHE_FLASH_ATTR, INFO, NULL, os_free, MQTT_Client::pCon, espconn::proto, and espconn::tcp.

Referenced by mqtt_client_delete(), MQTT_Connect(), and MQTT_Task().

{
    if (mqttClient->pCon != NULL) {
        INFO("TCP: Free memory\r\n");
        // Force abort connections
        espconn_abort(mqttClient->pCon);
        // Delete connections
        espconn_delete(mqttClient->pCon);
        
        if (mqttClient->pCon->proto.tcp) {
            os_free(mqttClient->pCon->proto.tcp);
            mqttClient->pCon->proto.tcp = NULL;
        }
        os_free(mqttClient->pCon);
        mqttClient->pCon = NULL;
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_tcpclient_discon_cb()

void ICACHE_FLASH_ATTR mqtt_tcpclient_discon_cb

(

void *

arg

)

Definition at line 498 of file mqtt.c.

References MQTT_Client::connState, MQTT_Client::disconnectedCb, ICACHE_FLASH_ATTR, INFO, MQTT_DELETED, MQTT_DELETING, MQTT_TASK_PRIO, os_param_t, espconn::reverse, system_os_post(), TCP_DISCONNECTED, TCP_DISCONNECTING, and TCP_RECONNECT_REQ.

Referenced by mqtt_tcpclient_connect_cb().

{

    struct espconn *pespconn = (struct espconn *)arg;
    MQTT_Client* client = (MQTT_Client *)pespconn->reverse;
    INFO("TCP: Disconnected callback\r\n");
    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);
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_tcpclient_recon_cb()

void ICACHE_FLASH_ATTR mqtt_tcpclient_recon_cb	(	void *	arg,
		sint8	errType
	)

Tcp client connect repeat callback function.

Parameters

arg	contain the ip link information

Return values

None

Definition at line 567 of file mqtt.c.

References BOOL, MQTT_Client::connState, MQTT_Client::host, ICACHE_FLASH_ATTR, INFO, MQTT_TASK_PRIO, os_param_t, MQTT_Client::port, espconn::reverse, system_os_post(), and TCP_RECONNECT_REQ.

Referenced by MQTT_Connect().

{
    struct espconn *pCon = (struct espconn *)arg;
    MQTT_Client* client = (MQTT_Client *)pCon->reverse;

    INFO("TCP: Reconnect to %s:%d\r\n", client->host, client->port);

    client->connState = TCP_RECONNECT_REQ;

    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);

}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_tcpclient_recv()

void ICACHE_FLASH_ATTR mqtt_tcpclient_recv	(	void *	arg,
		char *	pdata,
		unsigned short	len
	)

Client received callback function.

Parameters

arg	contain the ip link information
pdata	received data
len	the lenght of received data

Return values

None

Definition at line 296 of file mqtt.c.

References MQTT_Client::connectedCb, CONNECTION_ACCEPTED, CONNECTION_REFUSE_BAD_USERNAME, CONNECTION_REFUSE_NOT_AUTHORIZED, CONNECTION_REFUSE_PROTOCOL, CONNECTION_REFUSE_SERVER_UNAVAILABLE, MQTT_Client::connState, mqtt_message::data, deliver_publish(), espconn_disconnect(), espconn_secure_disconnect(), MQTT_Client::host, ICACHE_FLASH_ATTR, mqtt_state_t::in_buffer, mqtt_state_t::in_buffer_length, INFO, MQTT_Client::keepAliveTick, mqtt_message::length, mqtt_state_t::message_length, mqtt_state_t::message_length_read, MQTT_CONNECT_SENDING, mqtt_state_t::mqtt_connection, MQTT_DATA, mqtt_get_connect_return_code(), mqtt_get_id(), mqtt_get_qos(), mqtt_get_total_length(), mqtt_get_type(), MQTT_KEEPALIVE_SEND, mqtt_msg_pingresp(), mqtt_msg_puback(), mqtt_msg_pubcomp(), mqtt_msg_pubrec(), mqtt_msg_pubrel(), MQTT_MSG_TYPE_CONNACK, MQTT_MSG_TYPE_CONNECT, MQTT_MSG_TYPE_PINGREQ, MQTT_MSG_TYPE_PINGRESP, MQTT_MSG_TYPE_PUBACK, MQTT_MSG_TYPE_PUBCOMP, MQTT_MSG_TYPE_PUBLISH, MQTT_MSG_TYPE_PUBREC, MQTT_MSG_TYPE_PUBREL, MQTT_MSG_TYPE_SUBACK, MQTT_MSG_TYPE_SUBSCRIBE, MQTT_MSG_TYPE_UNSUBACK, MQTT_MSG_TYPE_UNSUBSCRIBE, MQTT_Client::mqtt_state, MQTT_TASK_PRIO, MQTT_Client::msgQueue, os_memcpy, os_param_t, mqtt_state_t::outbound_message, MQTT_Client::pCon, mqtt_state_t::pending_msg_id, mqtt_state_t::pending_msg_type, MQTT_Client::port, QUEUE_Puts(), espconn::reverse, MQTT_Client::security, and system_os_post().

Referenced by mqtt_tcpclient_connect_cb().

{
    uint8_t msg_type;
    uint8_t msg_qos;
    uint16_t msg_id;
    uint8_t msg_conn_ret;

    struct espconn *pCon = (struct espconn*)arg;
    MQTT_Client *client = (MQTT_Client *)pCon->reverse;

    client->keepAliveTick = 0;
READPACKET:
    INFO("TCP: data received %d bytes\r\n", len);
    // INFO("STATE: %d\r\n", client->connState);
    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) {
        case MQTT_CONNECT_SENDING:
            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) {
#ifdef MQTT_SSL_ENABLE
                        espconn_secure_disconnect(client->pCon);
#else
                        INFO("TCP: Do not support SSL\r\n");
#endif
                    }
                    else {
                        espconn_disconnect(client->pCon);
                    }
                } else {
                    msg_conn_ret = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
                    switch (msg_conn_ret) {
                        case CONNECTION_ACCEPTED:
                            INFO("MQTT: Connected to %s:%d\r\n", client->host, client->port);
                            client->connState = MQTT_DATA;
                            if (client->connectedCb)
                                client->connectedCb((uint32_t*)client);
                        break;
                        case CONNECTION_REFUSE_PROTOCOL:
                        case CONNECTION_REFUSE_SERVER_UNAVAILABLE:
                        case CONNECTION_REFUSE_BAD_USERNAME:
                        case CONNECTION_REFUSE_NOT_AUTHORIZED:
                            INFO("MQTT: Connection refuse, reason code: %d\r\n", msg_conn_ret);
                        default:
                        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);
                        }
                    }
                }

            }
            break;
        case MQTT_DATA:
        case MQTT_KEEPALIVE_SEND:
            client->mqtt_state.message_length_read = len;
            client->mqtt_state.message_length = mqtt_get_total_length(client->mqtt_state.in_buffer, client->mqtt_state.message_length_read);


            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");
                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");
                break;
            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)
                    client->mqtt_state.outbound_message = mqtt_msg_pubrec(&client->mqtt_state.mqtt_connection, msg_id);
                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) {
                        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");
                }

                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) {
                    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) {
                    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)
            {
                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;

                    INFO("Get another published message\r\n");
                    goto READPACKET;
                }

            }
            break;
        }
    } else {
        INFO("ERROR: Message too long\r\n");
    }
    system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_tcpclient_sent_cb()

void ICACHE_FLASH_ATTR mqtt_tcpclient_sent_cb

(

void *

arg

)

Client send over callback function.

Parameters

arg	contain the ip link information

Return values

None

Definition at line 456 of file mqtt.c.

References MQTT_Client::connState, INFO, MQTT_Client::keepAliveTick, MQTT_DATA, MQTT_KEEPALIVE_SEND, MQTT_MSG_TYPE_PUBLISH, MQTT_Client::mqtt_state, MQTT_TASK_PRIO, os_param_t, mqtt_state_t::pending_msg_type, MQTT_Client::publishedCb, espconn::reverse, MQTT_Client::sendTimeout, and system_os_post().

Referenced by mqtt_tcpclient_connect_cb().

{
    struct espconn *pCon = (struct espconn *)arg;
    MQTT_Client* client = (MQTT_Client *)pCon->reverse;
    INFO("TCP: Sent\r\n");
    client->sendTimeout = 0;
    client->keepAliveTick =0;

    if ((client->connState == MQTT_DATA || client->connState == MQTT_KEEPALIVE_SEND)
                && 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);
}

Here is the call graph for this function:

Here is the caller graph for this function:

mqtt_timer()

void ICACHE_FLASH_ATTR mqtt_timer

(

void *

arg

)

Definition at line 472 of file mqtt.c.

References mqtt_state_t::connect_info, MQTT_Client::connState, ICACHE_FLASH_ATTR, mqtt_connect_info::keepalive, MQTT_Client::keepAliveTick, MQTT_DATA, MQTT_KEEPALIVE_SEND, MQTT_RECONNECT_TIMEOUT, MQTT_Client::mqtt_state, MQTT_TASK_PRIO, os_param_t, MQTT_Client::reconnectTick, MQTT_Client::sendTimeout, system_os_post(), TCP_RECONNECT, and TCP_RECONNECT_REQ.

Referenced by MQTT_Connect().

{
    MQTT_Client* client = (MQTT_Client*)arg;

    if (client->connState == MQTT_DATA) {
        client->keepAliveTick ++;
        if (client->keepAliveTick > (client->mqtt_state.connect_info->keepalive / 2)) {
            client->connState = MQTT_KEEPALIVE_SEND;
            system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
        }

    } else if (client->connState == TCP_RECONNECT_REQ) {
        client->reconnectTick ++;
        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)
        client->sendTimeout --;
}

Here is the call graph for this function:

Here is the caller graph for this function:

MQTT_UnSubscribe()

BOOL ICACHE_FLASH_ATTR MQTT_UnSubscribe	(	MQTT_Client *	client,
		char *	topic
	)

MQTT un-subscibe function.

Parameters

client	MQTT_Client reference
topic	String topic will un-subscribe

Return values

TRUE	if success queue

Definition at line 650 of file mqtt.c.

References BOOL, mqtt_message::data, FALSE, ICACHE_FLASH_ATTR, INFO, mqtt_message::length, MQTT_BUF_SIZE, mqtt_state_t::mqtt_connection, mqtt_msg_unsubscribe(), MQTT_Client::mqtt_state, MQTT_TASK_PRIO, MQTT_Client::msgQueue, os_param_t, mqtt_state_t::outbound_message, mqtt_state_t::pending_msg_id, QUEUE_Gets(), QUEUE_Puts(), system_os_post(), and TRUE.

{
    uint8_t dataBuffer[MQTT_BUF_SIZE];
    uint16_t dataLen;
    client->mqtt_state.outbound_message = mqtt_msg_unsubscribe(&client->mqtt_state.mqtt_connection,
                                          topic,
                                          &client->mqtt_state.pending_msg_id);
    INFO("MQTT: queue un-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;
}

Here is the call graph for this function:

Variable Documentation

default_certificate

unsigned char* default_certificate

Definition at line 64 of file mqtt.c.

default_certificate_len

unsigned int default_certificate_len = 0

Definition at line 65 of file mqtt.c.

default_private_key

unsigned char* default_private_key

Definition at line 66 of file mqtt.c.

default_private_key_len

unsigned int default_private_key_len = 0

Definition at line 67 of file mqtt.c.

mqtt_procTaskQueue

os_event_t mqtt_procTaskQueue[MQTT_TASK_QUEUE_SIZE]

Definition at line 69 of file mqtt.c.

Referenced by MQTT_InitClient().