mqtt.h File Reference

#include "mqtt_msg.h"
#include "user_interface.h"
#include "queue.h"

Include dependency graph for mqtt.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	mqtt_event_data_t
struct	mqtt_state_t
struct	MQTT_Client

Macros
#define	SEC_NONSSL   0
#define	SEC_SSL   1
#define	MQTT_FLAG_CONNECTED   1
#define	MQTT_FLAG_READY   2
#define	MQTT_FLAG_EXIT   4
#define	MQTT_EVENT_TYPE_NONE   0
#define	MQTT_EVENT_TYPE_CONNECTED   1
#define	MQTT_EVENT_TYPE_DISCONNECTED   2
#define	MQTT_EVENT_TYPE_SUBSCRIBED   3
#define	MQTT_EVENT_TYPE_UNSUBSCRIBED   4
#define	MQTT_EVENT_TYPE_PUBLISH   5
#define	MQTT_EVENT_TYPE_PUBLISHED   6
#define	MQTT_EVENT_TYPE_EXITED   7
#define	MQTT_EVENT_TYPE_PUBLISH_CONTINUATION   8

Typedefs
typedef struct mqtt_event_data_t	mqtt_event_data_t
typedef struct mqtt_state_t	mqtt_state_t
typedef void(*	MqttCallback) (uint32_t *args)
typedef void(*	MqttDataCallback) (uint32_t *args, const char *topic, uint32_t topic_len, const char *data, uint32_t lengh)

Enumerations
enum	tConnState {   WIFI_INIT, WIFI_CONNECTING, WIFI_CONNECTING_ERROR, WIFI_CONNECTED,   DNS_RESOLVE, TCP_DISCONNECTING, TCP_DISCONNECTED, TCP_RECONNECT_DISCONNECTING,   TCP_RECONNECT_REQ, TCP_RECONNECT, TCP_CONNECTING, TCP_CONNECTING_ERROR,   TCP_CONNECTED, MQTT_CONNECT_SEND, MQTT_CONNECT_SENDING, MQTT_SUBSCIBE_SEND,   MQTT_SUBSCIBE_SENDING, MQTT_DATA, MQTT_KEEPALIVE_SEND, MQTT_PUBLISH_RECV,   MQTT_PUBLISHING, MQTT_DELETING, MQTT_DELETED }

Functions
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_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)
	MQTT subscibe function. More...
BOOL ICACHE_FLASH_ATTR	MQTT_UnSubscribe (MQTT_Client *client, char *topic)
	MQTT un-subscibe function. More...
void ICACHE_FLASH_ATTR	MQTT_Connect (MQTT_Client *mqttClient)
	Begin connect to MQTT broker. More...
void ICACHE_FLASH_ATTR	MQTT_Disconnect (MQTT_Client *mqttClient)
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_Ping (MQTT_Client *client)
	MQTT ping function. More...

Macro Definition Documentation

MQTT_EVENT_TYPE_CONNECTED

#define MQTT_EVENT_TYPE_CONNECTED   1

Definition at line 123 of file mqtt.h.

MQTT_EVENT_TYPE_DISCONNECTED

#define MQTT_EVENT_TYPE_DISCONNECTED   2

Definition at line 124 of file mqtt.h.

MQTT_EVENT_TYPE_EXITED

#define MQTT_EVENT_TYPE_EXITED   7

Definition at line 129 of file mqtt.h.

MQTT_EVENT_TYPE_NONE

#define MQTT_EVENT_TYPE_NONE   0

Definition at line 122 of file mqtt.h.

MQTT_EVENT_TYPE_PUBLISH

#define MQTT_EVENT_TYPE_PUBLISH   5

Definition at line 127 of file mqtt.h.

MQTT_EVENT_TYPE_PUBLISH_CONTINUATION

#define MQTT_EVENT_TYPE_PUBLISH_CONTINUATION   8

Definition at line 130 of file mqtt.h.

MQTT_EVENT_TYPE_PUBLISHED

#define MQTT_EVENT_TYPE_PUBLISHED   6

Definition at line 128 of file mqtt.h.

MQTT_EVENT_TYPE_SUBSCRIBED

#define MQTT_EVENT_TYPE_SUBSCRIBED   3

Definition at line 125 of file mqtt.h.

MQTT_EVENT_TYPE_UNSUBSCRIBED

#define MQTT_EVENT_TYPE_UNSUBSCRIBED   4

Definition at line 126 of file mqtt.h.

MQTT_FLAG_CONNECTED

#define MQTT_FLAG_CONNECTED   1

Definition at line 118 of file mqtt.h.

MQTT_FLAG_EXIT

#define MQTT_FLAG_EXIT   4

Definition at line 120 of file mqtt.h.

MQTT_FLAG_READY

#define MQTT_FLAG_READY   2

Definition at line 119 of file mqtt.h.

SEC_NONSSL

#define SEC_NONSSL   0

Definition at line 115 of file mqtt.h.

SEC_SSL

#define SEC_SSL   1

Definition at line 116 of file mqtt.h.

Typedef Documentation

mqtt_event_data_t

typedef struct mqtt_event_data_t mqtt_event_data_t

mqtt_state_t

typedef struct mqtt_state_t mqtt_state_t

MqttCallback

typedef void(* MqttCallback) (uint32_t *args)

Definition at line 90 of file mqtt.h.

MqttDataCallback

typedef void(* MqttDataCallback) (uint32_t *args, const char *topic, uint32_t topic_len, const char *data, uint32_t lengh)

Definition at line 91 of file mqtt.h.

Enumeration Type Documentation

tConnState

enum tConnState

Enumerator
WIFI_INIT
WIFI_CONNECTING
WIFI_CONNECTING_ERROR
WIFI_CONNECTED
DNS_RESOLVE
TCP_DISCONNECTING
TCP_DISCONNECTED
TCP_RECONNECT_DISCONNECTING
TCP_RECONNECT_REQ
TCP_RECONNECT
TCP_CONNECTING
TCP_CONNECTING_ERROR
TCP_CONNECTED
MQTT_CONNECT_SEND
MQTT_CONNECT_SENDING
MQTT_SUBSCIBE_SEND
MQTT_SUBSCIBE_SENDING
MQTT_DATA
MQTT_KEEPALIVE_SEND
MQTT_PUBLISH_RECV
MQTT_PUBLISHING
MQTT_DELETING
MQTT_DELETED

Definition at line 64 of file mqtt.h.

             {
    WIFI_INIT,
    WIFI_CONNECTING,
    WIFI_CONNECTING_ERROR,
    WIFI_CONNECTED,
    DNS_RESOLVE,
    TCP_DISCONNECTING,
    TCP_DISCONNECTED,
    TCP_RECONNECT_DISCONNECTING,
    TCP_RECONNECT_REQ,
    TCP_RECONNECT,
    TCP_CONNECTING,
    TCP_CONNECTING_ERROR,
    TCP_CONNECTED,
    MQTT_CONNECT_SEND,
    MQTT_CONNECT_SENDING,
    MQTT_SUBSCIBE_SEND,
    MQTT_SUBSCIBE_SENDING,
    MQTT_DATA,
    MQTT_KEEPALIVE_SEND,
    MQTT_PUBLISH_RECV,
    MQTT_PUBLISHING,
    MQTT_DELETING,
    MQTT_DELETED,
} tConnState;

Function Documentation

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_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_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_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: