diff --git a/Makefile b/Makefile index a3df665..7969417 100644 --- a/Makefile +++ b/Makefile @@ -142,7 +142,7 @@ TARGET = httpd APPGEN_TOOL ?= gen_appbin.py # which modules (subdirectories) of the project to include in compiling -MODULES = espfs httpd user serial cmd +MODULES = espfs httpd user serial cmd mqtt EXTRA_INCDIR = include . # libraries used in this project, mainly provided by the SDK diff --git a/esp-link.vcxproj b/esp-link.vcxproj index c98faa5..eac1d5b 100644 --- a/esp-link.vcxproj +++ b/esp-link.vcxproj @@ -28,7 +28,7 @@ __ets__;_STDINT_H;ICACHE_FLASH;__MINGW32__;__WIN32__ - .\cmd;.\serial;.\user;.\espfs;.\httpd;.\include;..\esp_iot_sdk_v1.3.0\include;..\xtensa-lx106-elf\xtensa-lx106-elf\include;c:\tools\mingw64\x86_64-w64-mingw32\include;c:\tools\mingw64\lib\gcc\x86_64-w64-mingw32\4.8.3\include + .\mqtt\include;.\cmd;.\serial;.\user;.\espfs;.\httpd;.\include;..\esp_iot_sdk_v1.3.0\include;..\xtensa-lx106-elf\xtensa-lx106-elf\include;c:\tools\mingw64\x86_64-w64-mingw32\include;c:\tools\mingw64\lib\gcc\x86_64-w64-mingw32\4.8.3\include @@ -75,6 +75,12 @@ + + + + + + @@ -106,6 +112,13 @@ + + + + + + + diff --git a/include/user_config.h b/include/user_config.h index 8b13789..d38c637 100644 --- a/include/user_config.h +++ b/include/user_config.h @@ -1 +1,39 @@ +#ifndef _USER_CONFIG_H_ +#define _USER_CONFIG_H_ +/*DEFAULT CONFIGURATIONS*/ + +#define MQTT_HOST "mqtt.yourdomain.com" //or "mqtt.yourdomain.com" +#define MQTT_PORT 1883 +#define MQTT_BUF_SIZE 1024 +#define MQTT_KEEPALIVE 120 /*second*/ + +#define MQTT_CLIENT_ID "H_%08X" //Cuidar para não colocar valores execendentes da ESTRUTURA SYSCFG +#define MQTT_USER "DVES_USER" +#define MQTT_PASS "DVES_PASS" + +#define STA_SSID "TESTE" +#define STA_PASS "54545" +#define STA_TYPE AUTH_WPA2_PSK + +#define MQTT_RECONNECT_TIMEOUT 5 /*second*/ + +#define DEFAULT_SECURITY 0 +#define QUEUE_BUFFER_SIZE 2048 + +//#undef MCU_RESET_PIN +//#undef MCU_ISP_PIN +//#undef LED_CONN_PIN +//#undef LED_SERIAL_PIN +// +//#define MCU_RESET_PIN 2 +//#define MCU_ISP_PIN -1 +//#define LED_CONN_PIN -1 +//#define LED_SERIAL_PIN -1 + +//#define BAUD_RATE 9600 +//#define HOSTNAME "nodemcu\0 " + +#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 \ No newline at end of file diff --git a/mqtt/include/mqtt.h b/mqtt/include/mqtt.h new file mode 100644 index 0000000..52d21fe --- /dev/null +++ b/mqtt/include/mqtt.h @@ -0,0 +1,140 @@ +/* mqtt.h +* +* Copyright (c) 2014-2015, Tuan PM +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions are met: +* +* * Redistributions of source code must retain the above copyright notice, +* this list of conditions and the following disclaimer. +* * Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* * Neither the name of Redis nor the names of its contributors may be used +* to endorse or promote products derived from this software without +* specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +*/ +#ifndef USER_AT_MQTT_H_ +#define USER_AT_MQTT_H_ + +#include +#include "mqtt_msg.h" +#include "queue.h" +#include "utils.h" + +typedef struct mqtt_event_data_t +{ + uint8_t type; + const char* topic; + const char* data; + uint16_t topic_length; + uint16_t data_length; + uint16_t data_offset; +} mqtt_event_data_t; + +typedef struct mqtt_state_t +{ + uint16_t port; + int auto_reconnect; + mqtt_connect_info_t* connect_info; + uint8_t* in_buffer; + uint8_t* out_buffer; + int in_buffer_length; + int out_buffer_length; + uint16_t message_length; + uint16_t message_length_read; + mqtt_message_t* outbound_message; + mqtt_connection_t mqtt_connection; + uint16_t pending_msg_id; + int pending_msg_type; + int pending_publish_qos; +} mqtt_state_t; + +typedef enum { + WIFI_INIT, + WIFI_CONNECTING, + WIFI_CONNECTING_ERROR, + WIFI_CONNECTED, + DNS_RESOLVE, + TCP_DISCONNECTED, + 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_PUBLISH_RECV, + MQTT_PUBLISHING +} tConnState; + +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); + +typedef struct { + struct espconn *pCon; + uint8_t security; + uint8_t* host; + uint32_t port; + ip_addr_t ip; + mqtt_state_t mqtt_state; + mqtt_connect_info_t connect_info; + MqttCallback connectedCb; + MqttCallback disconnectedCb; + MqttCallback publishedCb; + MqttDataCallback dataCb; + ETSTimer mqttTimer; + uint32_t keepAliveTick; + uint32_t reconnectTick; + uint32_t sendTimeout; + tConnState connState; + QUEUE msgQueue; + void* user_data; +} MQTT_Client; + +#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 + +void ICACHE_FLASH_ATTR MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32 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_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_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); +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 qos, int retain); + +#endif /* USER_AT_MQTT_H_ */ diff --git a/mqtt/include/mqtt_msg.h b/mqtt/include/mqtt_msg.h new file mode 100644 index 0000000..f16bb80 --- /dev/null +++ b/mqtt/include/mqtt_msg.h @@ -0,0 +1,129 @@ +/* + * File: mqtt_msg.h + * Author: Minh Tuan + * + * Created on July 12, 2014, 1:05 PM + */ + +#ifndef MQTT_MSG_H +#define MQTT_MSG_H +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/* +* Copyright (c) 2014, Stephen Robinson +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions +* are met: +* +* 1. Redistributions of source code must retain the above copyright +* notice, this list of conditions and the following disclaimer. +* 2. Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* 3. Neither the name of the copyright holder nor the names of its +* contributors may be used to endorse or promote products derived +* from this software without specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +* +*/ +/* 7 6 5 4 3 2 1 0*/ +/* | --- Message Type ---- | DUP Flag | QoS Level | Retain | Remaining Length | */ + + +enum mqtt_message_type +{ + MQTT_MSG_TYPE_CONNECT = 1, + MQTT_MSG_TYPE_CONNACK = 2, + MQTT_MSG_TYPE_PUBLISH = 3, + MQTT_MSG_TYPE_PUBACK = 4, + MQTT_MSG_TYPE_PUBREC = 5, + MQTT_MSG_TYPE_PUBREL = 6, + MQTT_MSG_TYPE_PUBCOMP = 7, + MQTT_MSG_TYPE_SUBSCRIBE = 8, + MQTT_MSG_TYPE_SUBACK = 9, + MQTT_MSG_TYPE_UNSUBSCRIBE = 10, + MQTT_MSG_TYPE_UNSUBACK = 11, + MQTT_MSG_TYPE_PINGREQ = 12, + MQTT_MSG_TYPE_PINGRESP = 13, + MQTT_MSG_TYPE_DISCONNECT = 14 +}; + +typedef struct mqtt_message +{ + uint8_t* data; + uint16_t length; + +} mqtt_message_t; + +typedef struct mqtt_connection +{ + mqtt_message_t message; + + uint16_t message_id; + uint8_t* buffer; + uint16_t buffer_length; + +} mqtt_connection_t; + +typedef struct mqtt_connect_info +{ + char* client_id; + char* username; + char* password; + char* will_topic; + char* will_message; + int keepalive; + int will_qos; + int will_retain; + int clean_session; + +} mqtt_connect_info_t; + + +static inline int ICACHE_FLASH_ATTR mqtt_get_type(uint8_t* buffer) { return (buffer[0] & 0xf0) >> 4; } +static inline int ICACHE_FLASH_ATTR mqtt_get_dup(uint8_t* buffer) { return (buffer[0] & 0x08) >> 3; } +static inline int ICACHE_FLASH_ATTR mqtt_get_qos(uint8_t* buffer) { return (buffer[0] & 0x06) >> 1; } +static inline int ICACHE_FLASH_ATTR mqtt_get_retain(uint8_t* buffer) { return (buffer[0] & 0x01); } + +void ICACHE_FLASH_ATTR mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length); +int ICACHE_FLASH_ATTR mqtt_get_total_length(uint8_t* buffer, uint16_t length); +const char* ICACHE_FLASH_ATTR mqtt_get_publish_topic(uint8_t* buffer, uint16_t* length); +const char* ICACHE_FLASH_ATTR mqtt_get_publish_data(uint8_t* buffer, uint16_t* length); +uint16_t ICACHE_FLASH_ATTR mqtt_get_id(uint8_t* buffer, uint16_t length); + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pingreq(mqtt_connection_t* connection); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pingresp(mqtt_connection_t* connection); +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_disconnect(mqtt_connection_t* connection); + + +#ifdef __cplusplus +} +#endif + +#endif /* MQTT_MSG_H */ + diff --git a/mqtt/include/proto.h b/mqtt/include/proto.h new file mode 100644 index 0000000..73a4e8b --- /dev/null +++ b/mqtt/include/proto.h @@ -0,0 +1,31 @@ +/* + * File: proto.h + * Author: ThuHien + * + * Created on November 23, 2012, 8:57 AM + */ + +#ifndef _PROTO_H_ +#define _PROTO_H_ +#include +#include "ringbuf.h" + +typedef void(PROTO_PARSE_CALLBACK)(); + +typedef struct{ + uint8_t *buf; + uint16_t bufSize; + uint16_t dataLen; + uint8_t isEsc; + uint8_t isBegin; + PROTO_PARSE_CALLBACK* callback; +}PROTO_PARSER; + +int8_t ICACHE_FLASH_ATTR PROTO_Init(PROTO_PARSER *parser, PROTO_PARSE_CALLBACK *completeCallback, uint8_t *buf, uint16_t bufSize); +int8_t ICACHE_FLASH_ATTR PROTO_Parse(PROTO_PARSER *parser, uint8_t *buf, uint16_t len); +int16_t ICACHE_FLASH_ATTR PROTO_Add(uint8_t *buf, const uint8_t *packet, int16_t bufSize); +int16_t ICACHE_FLASH_ATTR PROTO_AddRb(RINGBUF *rb, const uint8_t *packet, int16_t len); +int8_t ICACHE_FLASH_ATTR PROTO_ParseByte(PROTO_PARSER *parser, uint8_t value); +int16_t ICACHE_FLASH_ATTR PROTO_ParseRb(RINGBUF *rb, uint8_t *bufOut, uint16_t* len, uint16_t maxBufLen); +#endif + diff --git a/mqtt/include/queue.h b/mqtt/include/queue.h new file mode 100644 index 0000000..7a84480 --- /dev/null +++ b/mqtt/include/queue.h @@ -0,0 +1,45 @@ +/* str_queue.h -- +* +* Copyright (c) 2014-2015, Tuan PM +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions are met: +* +* * Redistributions of source code must retain the above copyright notice, +* this list of conditions and the following disclaimer. +* * Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* * Neither the name of Redis nor the names of its contributors may be used +* to endorse or promote products derived from this software without +* specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +*/ + +#ifndef USER_QUEUE_H_ +#define USER_QUEUE_H_ +#include +#include "proto.h" +#include "ringbuf.h" +typedef struct { + uint8_t *buf; + RINGBUF rb; +} QUEUE; + +void ICACHE_FLASH_ATTR QUEUE_Init(QUEUE *queue, int bufferSize); +int32_t ICACHE_FLASH_ATTR QUEUE_Puts(QUEUE *queue, uint8_t* buffer, uint16_t len); +int32_t ICACHE_FLASH_ATTR QUEUE_Gets(QUEUE *queue, uint8_t* buffer, uint16_t* len, uint16_t maxLen); +bool ICACHE_FLASH_ATTR QUEUE_IsEmpty(QUEUE *queue); +#endif /* USER_QUEUE_H_ */ diff --git a/mqtt/include/ringbuf.h b/mqtt/include/ringbuf.h new file mode 100644 index 0000000..03d780d --- /dev/null +++ b/mqtt/include/ringbuf.h @@ -0,0 +1,18 @@ +#ifndef _RING_BUF_H_ +#define _RING_BUF_H_ + +#include +#include "typedef.h" + +typedef struct{ + U8* p_o; /**< Original pointer */ + U8* volatile p_r; /**< Read pointer */ + U8* volatile p_w; /**< Write pointer */ + volatile I32 fill_cnt; /**< Number of filled slots */ + I32 size; /**< Buffer size */ +}RINGBUF; + +I16 ICACHE_FLASH_ATTR RINGBUF_Init(RINGBUF *r, U8* buf, I32 size); +I16 ICACHE_FLASH_ATTR RINGBUF_Put(RINGBUF *r, U8 c); +I16 ICACHE_FLASH_ATTR RINGBUF_Get(RINGBUF *r, U8* c); +#endif diff --git a/mqtt/include/typedef.h b/mqtt/include/typedef.h new file mode 100644 index 0000000..a4c69d6 --- /dev/null +++ b/mqtt/include/typedef.h @@ -0,0 +1,17 @@ +/** +* \file +* Standard Types definition +*/ + +#ifndef _TYPE_DEF_H_ +#define _TYPE_DEF_H_ + +typedef char I8; +typedef unsigned char U8; +typedef short I16; +typedef unsigned short U16; +typedef long I32; +typedef unsigned long U32; +typedef unsigned long long U64; + +#endif diff --git a/mqtt/include/utils.h b/mqtt/include/utils.h new file mode 100644 index 0000000..676501e --- /dev/null +++ b/mqtt/include/utils.h @@ -0,0 +1,9 @@ +#ifndef _UTILS_H_ +#define _UTILS_H_ + +#include + +uint32_t ICACHE_FLASH_ATTR UTILS_Atoh(const int8_t *s); +uint8_t ICACHE_FLASH_ATTR UTILS_StrToIP(const int8_t* str, void *ip); +uint8_t ICACHE_FLASH_ATTR UTILS_IsIPV4 (int8_t *str); +#endif diff --git a/mqtt/mqtt.c b/mqtt/mqtt.c new file mode 100644 index 0000000..49ed507 --- /dev/null +++ b/mqtt/mqtt.c @@ -0,0 +1,637 @@ +/* mqtt.c +* Protocol: http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html +* +* Copyright (c) 2014-2015, Tuan PM +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions are met: +* +* * Redistributions of source code must retain the above copyright notice, +* this list of conditions and the following disclaimer. +* * Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* * Neither the name of Redis nor the names of its contributors may be used +* to endorse or promote products derived from this software without +* specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +*/ + +#include "mqtt.h" + +#define MQTT_TASK_PRIO 0 +#define MQTT_TASK_QUEUE_SIZE 1 +#define MQTT_SEND_TIMOUT 5 + +#ifndef QUEUE_BUFFER_SIZE +#define QUEUE_BUFFER_SIZE 2048 +#endif + +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]; + +LOCAL void ICACHE_FLASH_ATTR +mqtt_dns_found(const char *name, ip_addr_t *ipaddr, void *arg) +{ + struct espconn *pConn = (struct espconn *)arg; + MQTT_Client* client = (MQTT_Client *)pConn->reverse; + + + if (ipaddr == NULL) + { + os_printf("DNS: Found, but got no ip, try to reconnect\n"); + client->connState = TCP_RECONNECT_REQ; + return; + } + + os_printf("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); +#ifdef CLIENT_SSL_ENABLE + if (client->security){ + espconn_secure_connect(client->pCon); + } + else +#endif + espconn_connect(client->pCon); + + client->connState = TCP_CONNECTING; + os_printf("MQTT-TCP: connecting...\n"); + } + + system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); +} + + + +LOCAL void ICACHE_FLASH_ATTR +deliver_publish(MQTT_Client* client, uint8_t* message, int length) +{ + 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); + +} + + +/** +* @brief Client received callback function. +* @param arg: contain the ip link information +* @param pdata: received data +* @param len: the lenght of received data +* @retval None +*/ +void ICACHE_FLASH_ATTR +mqtt_tcpclient_recv(void *arg, char *pdata, unsigned short len) +{ + uint8_t msg_type; + uint8_t msg_qos; + uint16_t msg_id; + + struct espconn *pCon = (struct espconn*)arg; + MQTT_Client *client = (MQTT_Client *)pCon->reverse; + +READPACKET: + os_printf("MQTT-TCP: Data received %d bytes\n", len); + 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); + if (client->connState == MQTT_CONNECT_SENDING) { + if (msg_type == MQTT_MSG_TYPE_CONNACK){ + if (client->mqtt_state.pending_msg_type != MQTT_MSG_TYPE_CONNECT){ + os_printf("MQTT: Invalid packet\n"); +#ifdef CLIENT_SSL_ENABLE + if (client->security){ + espconn_secure_disconnect(client->pCon); + } + else +#endif + espconn_disconnect(client->pCon); + } + else { + os_printf("MQTT: Connected to %s:%ld\n", client->host, client->port); + client->connState = MQTT_DATA; + if (client->connectedCb) + client->connectedCb((uint32_t*)client); + } + } + } + else if (client->connState == MQTT_DATA) { + 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); + + + if (msg_type == MQTT_MSG_TYPE_SUBACK) { + if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_SUBSCRIBE && client->mqtt_state.pending_msg_id == msg_id) + os_printf("MQTT: Subscribe successful\n"); + } + else if (msg_type == MQTT_MSG_TYPE_UNSUBACK){ + if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_UNSUBSCRIBE && client->mqtt_state.pending_msg_id == msg_id) + os_printf("MQTT: UnSubscribe successful\n"); + } + else if (msg_type == 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){ + os_printf("MQTT: Queue response QoS: %d\n", msg_qos); + if (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1){ + os_printf("MQTT: Queue full\n"); + } + } + deliver_publish(client, client->mqtt_state.in_buffer, client->mqtt_state.message_length_read); + } + else if (msg_type == MQTT_MSG_TYPE_PUBACK) { + if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_msg_id == msg_id){ + os_printf("MQTT: received MQTT_MSG_TYPE_PUBACK, finish QoS1 publish\n"); + } + } + else if (msg_type == 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){ + os_printf("MQTT: Queue full\n"); + } + } + else if (msg_type == 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){ + os_printf("MQTT: Queue full\n"); + } + } + else if (msg_type == MQTT_MSG_TYPE_PUBCOMP) { + if (client->mqtt_state.pending_msg_type == MQTT_MSG_TYPE_PUBLISH && client->mqtt_state.pending_msg_id == msg_id){ + os_printf("MQTT: receive MQTT_MSG_TYPE_PUBCOMP, finish QoS2 publish\n"); + } + } + else if (msg_type == 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){ + os_printf("MQTT: Queue full\n"); + } + } + + // 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; + + os_printf("Get another published message\n"); + goto READPACKET; + } + } + } + } + else { + os_printf("ERROR: Message too long\n"); + } + system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); +} + +/** +* @brief Client send over callback function. +* @param arg: contain the ip link information +* @retval None +*/ +void ICACHE_FLASH_ATTR +mqtt_tcpclient_sent_cb(void *arg) +{ + struct espconn *pCon = (struct espconn *)arg; + MQTT_Client* client = (MQTT_Client *)pCon->reverse; + os_printf("MQTT-TCP: Sent\n"); + client->sendTimeout = 0; + 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); +} + +void ICACHE_FLASH_ATTR mqtt_timer(void *arg) +{ + MQTT_Client* client = (MQTT_Client*)arg; + + if (client->connState == MQTT_DATA){ + client->keepAliveTick++; + if (client->keepAliveTick > client->mqtt_state.connect_info->keepalive){ + + os_printf("\nMQTT: Send keepalive packet to %s:%ld!\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; + os_printf("MQTT: Sending, type: %d, id: %04X\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id); +#ifdef CLIENT_SSL_ENABLE + if (client->security){ + espconn_secure_sent(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length); + } + else +#endif + espconn_sent(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); + } + + } + 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->sendTimeout > 0) + client->sendTimeout--; +} + +void ICACHE_FLASH_ATTR +mqtt_tcpclient_discon_cb(void *arg) +{ + + struct espconn *pespconn = (struct espconn *)arg; + MQTT_Client* client = (MQTT_Client *)pespconn->reverse; + os_printf("MQTT-TCP: Disconnected callback\n"); + client->connState = TCP_RECONNECT_REQ; + if (client->disconnectedCb) + client->disconnectedCb((uint32_t*)client); + + system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); +} + +/** +* @brief Tcp client connect success callback function. +* @param arg: contain the ip link information +* @retval None +*/ +void ICACHE_FLASH_ATTR +mqtt_tcpclient_connect_cb(void *arg) +{ + 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);/////// + os_printf("MQTT: Connected to broker %s:%ld\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; + os_printf("MQTT: Sending, type: %d, id: %04X\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id); +#ifdef CLIENT_SSL_ENABLE + if (client->security){ + espconn_secure_sent(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length); + } + else +#endif + espconn_sent(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); +} + +/** +* @brief Tcp client connect repeat callback function. +* @param arg: contain the ip link information +* @retval None +*/ +void ICACHE_FLASH_ATTR +mqtt_tcpclient_recon_cb(void *arg, sint8 errType) +{ + struct espconn *pCon = (struct espconn *)arg; + MQTT_Client* client = (MQTT_Client *)pCon->reverse; + + os_printf("MQTT-TCP: Reconnect to %s:%ld\n", client->host, client->port); + + client->connState = TCP_RECONNECT_REQ; + + system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); +} + +/** +* @brief MQTT publish function. +* @param client: MQTT_Client reference +* @param topic: string topic will publish to +* @param data: buffer data send point to +* @param data_length: length of data +* @param qos: qos +* @param retain: retain +* @retval TRUE if success queue +*/ +bool ICACHE_FLASH_ATTR +MQTT_Publish(MQTT_Client *client, const char* topic, const char* data, int qos, int retain) +{ + int data_length = os_strlen(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){ + os_printf("MQTT: Queuing Publish failed\n"); + return FALSE; + } + os_printf("MQTT: Queuing Publish, length: %d, queue size(%ld/%ld)\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){ + os_printf("MQTT: Queue full\n"); + if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) { + os_printf("MQTT: Serious buffer error\n"); + return FALSE; + } + } + system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); + return TRUE; +} + +/** +* @brief MQTT subscibe function. +* @param client: MQTT_Client reference +* @param topic: string topic will subscribe +* @param qos: qos +* @retval TRUE if success queue +*/ +bool ICACHE_FLASH_ATTR +MQTT_Subscribe(MQTT_Client *client, char* topic, uint8_t qos) +{ + uint8_t dataBuffer[MQTT_BUF_SIZE]; + uint16_t dataLen; + + client->mqtt_state.outbound_message = mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection, + topic, 0, + &client->mqtt_state.pending_msg_id); + os_printf("MQTT: Queue Subscribe, topic: \"%s\", id: %d\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){ + os_printf("MQTT: Queue full\n"); + if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) { + os_printf("MQTT: Serious buffer error\n"); + return FALSE; + } + } + system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); + return TRUE; +} + +void ICACHE_FLASH_ATTR 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) + return; + + if (client->connState == TCP_RECONNECT_REQ) { + return; + } + else if (client->connState == TCP_RECONNECT){ + MQTT_Connect(client); + os_printf("MQTT-TCP: Reconnect to: %s:%ld\n", client->host, client->port); + client->connState = TCP_CONNECTING; + } + else if (client->connState == MQTT_DATA) { + if (QUEUE_IsEmpty(&client->msgQueue) || client->sendTimeout != 0) + return; + + 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; + os_printf("MQTT: Sending, type: %d, id: %04X\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id); +#ifdef CLIENT_SSL_ENABLE + if (client->security){ + espconn_secure_sent(client->pCon, dataBuffer, dataLen); + } + else +#endif + espconn_sent(client->pCon, dataBuffer, dataLen); + + client->mqtt_state.outbound_message = NULL; + return; + } + return; + } +} + +/** +* @brief MQTT initialization connection function +* @param client: MQTT_Client reference +* @param host: Domain or IP string +* @param port: Port to connect +* @param security: 1 for ssl, 0 for none +* @retval None +*/ +void ICACHE_FLASH_ATTR +MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32 port, uint8_t security) +{ + uint32_t temp; + os_printf("MQTT_InitConnection\n"); + os_memset(mqttClient, 0, sizeof(MQTT_Client)); + temp = sizeof((char*)host); + mqttClient->host = (uint8_t*)os_zalloc(temp + 1); + os_strcpy((char*)mqttClient->host, (char*)host); + mqttClient->host[temp] = 0; + mqttClient->port = port; + mqttClient->security = security; +} + +/** +* @brief MQTT initialization mqtt client function +* @param client: MQTT_Client reference +* @param clientid: MQTT client id +* @param client_user:MQTT client user +* @param client_pass:MQTT client password +* @param client_pass:MQTT keep alive timer, in second +* @retval None +*/ +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) +{ + uint32_t temp; + os_printf("MQTT_InitClient\n"); + + os_memset(&mqttClient->connect_info, 0, sizeof(mqtt_connect_info_t)); + + temp = os_strlen((char*)client_id); + mqttClient->connect_info.client_id = (char*)os_zalloc(temp + 1); + os_strcpy((char*)mqttClient->connect_info.client_id, (char*)client_id); + mqttClient->connect_info.client_id[temp] = 0; + + temp = os_strlen((char*)client_user); + mqttClient->connect_info.username = (char*)os_zalloc(temp + 1); + os_strcpy((char*)mqttClient->connect_info.username, (char*)client_user); + mqttClient->connect_info.username[temp] = 0; + + temp = os_strlen((char*)client_pass); + mqttClient->connect_info.password = (char*)os_zalloc(temp + 1); + os_strcpy((char*)mqttClient->connect_info.password, (char*)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); +} + +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) +{ + uint32_t temp; + temp = os_strlen((char*)will_topic); + mqttClient->connect_info.will_topic = (char*)os_zalloc(temp + 1); + os_strcpy((char*)mqttClient->connect_info.will_topic, (char*)will_topic); + mqttClient->connect_info.will_topic[temp] = 0; + + temp = os_strlen((char*)will_msg); + mqttClient->connect_info.will_message = (char*)os_zalloc(temp + 1); + os_strcpy((char*)mqttClient->connect_info.will_message, (char*)will_msg); + mqttClient->connect_info.will_message[temp] = 0; + + + mqttClient->connect_info.will_qos = will_qos; + mqttClient->connect_info.will_retain = will_retain; +} + +/** +* @brief Begin connect to MQTT broker +* @param client: MQTT_Client reference +* @retval None +*/ +void ICACHE_FLASH_ATTR +MQTT_Connect(MQTT_Client *mqttClient) +{ + MQTT_Disconnect(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; + 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((const int8_t *)mqttClient->host, (void*)&mqttClient->pCon->proto.tcp->remote_ip)) { + os_printf("MQTT-TCP: Connect to ip %s:%ld\n", mqttClient->host, mqttClient->port); +#ifdef CLIENT_SSL_ENABLE + if (mqttClient->security){ + espconn_secure_connect(mqttClient->pCon); + } + else +#endif + espconn_connect(mqttClient->pCon); + } + else { + os_printf("MQTT-TCP: Connect to domain %s:%ld\n", mqttClient->host, mqttClient->port); + espconn_gethostbyname(mqttClient->pCon, (const char *)mqttClient->host, &mqttClient->ip, mqtt_dns_found); + } + mqttClient->connState = TCP_CONNECTING; +} + +void ICACHE_FLASH_ATTR +MQTT_Disconnect(MQTT_Client *mqttClient) +{ + if (mqttClient->pCon){ + os_printf("Free memory\n"); + if (mqttClient->pCon->proto.tcp) + os_free(mqttClient->pCon->proto.tcp); + os_free(mqttClient->pCon); + mqttClient->pCon = NULL; + } + + os_timer_disarm(&mqttClient->mqttTimer); +} + +void ICACHE_FLASH_ATTR +MQTT_OnConnected(MQTT_Client *mqttClient, MqttCallback connectedCb) +{ + mqttClient->connectedCb = connectedCb; +} + +void ICACHE_FLASH_ATTR +MQTT_OnDisconnected(MQTT_Client *mqttClient, MqttCallback disconnectedCb) +{ + mqttClient->disconnectedCb = disconnectedCb; +} + +void ICACHE_FLASH_ATTR +MQTT_OnData(MQTT_Client *mqttClient, MqttDataCallback dataCb) +{ + mqttClient->dataCb = dataCb; +} + +void ICACHE_FLASH_ATTR +MQTT_OnPublished(MQTT_Client *mqttClient, MqttCallback publishedCb) +{ + mqttClient->publishedCb = publishedCb; +} diff --git a/mqtt/mqtt_msg.c b/mqtt/mqtt_msg.c new file mode 100644 index 0000000..7e159c6 --- /dev/null +++ b/mqtt/mqtt_msg.c @@ -0,0 +1,471 @@ +/* +* Copyright (c) 2014, Stephen Robinson +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions +* are met: +* +* 1. Redistributions of source code must retain the above copyright +* notice, this list of conditions and the following disclaimer. +* 2. Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* 3. Neither the name of the copyright holder nor the names of its +* contributors may be used to endorse or promote products derived +* from this software without specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +* +*/ + +#include "mqtt_msg.h" +#define MQTT_MAX_FIXED_HEADER_SIZE 3 + +enum mqtt_connect_flag +{ + MQTT_CONNECT_FLAG_USERNAME = 1 << 7, + MQTT_CONNECT_FLAG_PASSWORD = 1 << 6, + MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5, + MQTT_CONNECT_FLAG_WILL = 1 << 2, + MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1 +}; + +struct __attribute__((__packed__)) mqtt_connect_variable_header +{ + uint8_t lengthMsb; + uint8_t lengthLsb; +#if defined(PROTOCOL_NAMEv31) + uint8_t magic[6]; +#elif defined(PROTOCOL_NAMEv311) + uint8_t magic[4]; +#else +#error "Please define protocol name" +#endif + uint8_t version; + uint8_t flags; + uint8_t keepaliveMsb; + uint8_t keepaliveLsb; +}; + +static int ICACHE_FLASH_ATTR append_string(mqtt_connection_t* connection, const char* string, int len) +{ + if (connection->message.length + len + 2 > connection->buffer_length) + return -1; + + connection->buffer[connection->message.length++] = len >> 8; + connection->buffer[connection->message.length++] = len & 0xff; + memcpy(connection->buffer + connection->message.length, string, len); + connection->message.length += len; + + return len + 2; +} + +static uint16_t ICACHE_FLASH_ATTR append_message_id(mqtt_connection_t* connection, uint16_t message_id) +{ + // If message_id is zero then we should assign one, otherwise + // we'll use the one supplied by the caller + while (message_id == 0) + message_id = ++connection->message_id; + + if (connection->message.length + 2 > connection->buffer_length) + return 0; + + connection->buffer[connection->message.length++] = message_id >> 8; + connection->buffer[connection->message.length++] = message_id & 0xff; + + return message_id; +} + +static int ICACHE_FLASH_ATTR init_message(mqtt_connection_t* connection) +{ + connection->message.length = MQTT_MAX_FIXED_HEADER_SIZE; + return MQTT_MAX_FIXED_HEADER_SIZE; +} + +static mqtt_message_t* ICACHE_FLASH_ATTR fail_message(mqtt_connection_t* connection) +{ + connection->message.data = connection->buffer; + connection->message.length = 0; + return &connection->message; +} + +static mqtt_message_t* ICACHE_FLASH_ATTR fini_message(mqtt_connection_t* connection, int type, int dup, int qos, int retain) +{ + int remaining_length = connection->message.length - MQTT_MAX_FIXED_HEADER_SIZE; + + if (remaining_length > 127) + { + connection->buffer[0] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1); + connection->buffer[1] = 0x80 | (remaining_length % 128); + connection->buffer[2] = remaining_length / 128; + connection->message.length = remaining_length + 3; + connection->message.data = connection->buffer; + } + else + { + connection->buffer[1] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1); + connection->buffer[2] = remaining_length; + connection->message.length = remaining_length + 2; + connection->message.data = connection->buffer + 1; + } + + return &connection->message; +} + +void ICACHE_FLASH_ATTR +mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length) { + memset(connection, 0, sizeof(connection)); + connection->buffer = buffer; + connection->buffer_length = buffer_length; +} + +int ICACHE_FLASH_ATTR mqtt_get_total_length(uint8_t* buffer, uint16_t length) +{ + int i; + int totlen = 0; + + for (i = 1; i < length; ++i) + { + totlen += (buffer[i] & 0x7f) << (7 * (i - 1)); + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + totlen += i; + + return totlen; +} + +const char* ICACHE_FLASH_ATTR mqtt_get_publish_topic(uint8_t* buffer, uint16_t* length) +{ + int i; + int totlen = 0; + int topiclen; + + for (i = 1; i < *length; ++i) + { + totlen += (buffer[i] & 0x7f) << (7 * (i - 1)); + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + totlen += i; + + if (i + 2 >= *length) + return NULL; + topiclen = buffer[i++] << 8; + topiclen |= buffer[i++]; + + if (i + topiclen > *length) + return NULL; + + *length = topiclen; + return (const char*)(buffer + i); +} + +const char* ICACHE_FLASH_ATTR mqtt_get_publish_data(uint8_t* buffer, uint16_t* length) +{ + int i; + int totlen = 0; + int topiclen; + + for (i = 1; i < *length; ++i) + { + totlen += (buffer[i] & 0x7f) << (7 * (i - 1)); + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + totlen += i; + + if (i + 2 >= *length) + return NULL; + topiclen = buffer[i++] << 8; + topiclen |= buffer[i++]; + + if (i + topiclen >= *length){ + *length = 0; + return NULL; + } + i += topiclen; + + if (mqtt_get_qos(buffer) > 0) + { + if (i + 2 >= *length) + return NULL; + i += 2; + } + + if (totlen < i) + return NULL; + + if (totlen <= *length) + *length = totlen - i; + else + *length = *length - i; + return (const char*)(buffer + i); +} + +uint16_t ICACHE_FLASH_ATTR mqtt_get_id(uint8_t* buffer, uint16_t length) +{ + if (length < 1) + return 0; + + switch (mqtt_get_type(buffer)) + { + case MQTT_MSG_TYPE_PUBLISH: + { + int i; + int topiclen; + + for (i = 1; i < length; ++i) + { + if ((buffer[i] & 0x80) == 0) + { + ++i; + break; + } + } + + if (i + 2 >= length) + return 0; + topiclen = buffer[i++] << 8; + topiclen |= buffer[i++]; + + if (i + topiclen >= length) + return 0; + i += topiclen; + + if (mqtt_get_qos(buffer) > 0) + { + if (i + 2 >= length) + return 0; + //i += 2; + } + else { + return 0; + } + + return (buffer[i] << 8) | buffer[i + 1]; + } + case MQTT_MSG_TYPE_PUBACK: + case MQTT_MSG_TYPE_PUBREC: + case MQTT_MSG_TYPE_PUBREL: + case MQTT_MSG_TYPE_PUBCOMP: + case MQTT_MSG_TYPE_SUBACK: + case MQTT_MSG_TYPE_UNSUBACK: + case MQTT_MSG_TYPE_SUBSCRIBE: + { + // This requires the remaining length to be encoded in 1 byte, + // which it should be. + if (length >= 4 && (buffer[1] & 0x80) == 0) + return (buffer[2] << 8) | buffer[3]; + else + return 0; + } + + default: + return 0; + } +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info) +{ + struct mqtt_connect_variable_header* variable_header; + + init_message(connection); + + if (connection->message.length + sizeof(*variable_header) > connection->buffer_length) + return fail_message(connection); + variable_header = (void*)(connection->buffer + connection->message.length); + connection->message.length += sizeof(*variable_header); + + variable_header->lengthMsb = 0; +#if defined(PROTOCOL_NAMEv31) + variable_header->lengthLsb = 6; + memcpy(variable_header->magic, "MQIsdp", 6); + variable_header->version = 3; +#elif defined(PROTOCOL_NAMEv311) + variable_header->lengthLsb = 4; + memcpy(variable_header->magic, "MQTT", 4); + variable_header->version = 4; +#else +#error "Please define protocol name" +#endif + + variable_header->flags = 0; + variable_header->keepaliveMsb = info->keepalive >> 8; + variable_header->keepaliveLsb = info->keepalive & 0xff; + + if (info->clean_session) + variable_header->flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION; + + if (info->client_id != NULL && info->client_id[0] != '\0') + { + if (append_string(connection, info->client_id, strlen(info->client_id)) < 0) + return fail_message(connection); + } + else + return fail_message(connection); + + if (info->will_topic != NULL && info->will_topic[0] != '\0') + { + if (append_string(connection, info->will_topic, strlen(info->will_topic)) < 0) + return fail_message(connection); + + if (append_string(connection, info->will_message, strlen(info->will_message)) < 0) + return fail_message(connection); + + variable_header->flags |= MQTT_CONNECT_FLAG_WILL; + if (info->will_retain) + variable_header->flags |= MQTT_CONNECT_FLAG_WILL_RETAIN; + variable_header->flags |= (info->will_qos & 3) << 3; + } + + if (info->username != NULL && info->username[0] != '\0') + { + if (append_string(connection, info->username, strlen(info->username)) < 0) + return fail_message(connection); + + variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME; + } + + if (info->password != NULL && info->password[0] != '\0') + { + if (append_string(connection, info->password, strlen(info->password)) < 0) + return fail_message(connection); + + variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD; + } + + return fini_message(connection, MQTT_MSG_TYPE_CONNECT, 0, 0, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_publish(mqtt_connection_t* connection, const char* topic, const char* data, int data_length, int qos, int retain, uint16_t* message_id) +{ + init_message(connection); + + if (topic == NULL || topic[0] == '\0') + return fail_message(connection); + + if (append_string(connection, topic, strlen(topic)) < 0) + return fail_message(connection); + + if (qos > 0) + { + if ((*message_id = append_message_id(connection, 0)) == 0) + return fail_message(connection); + } + else + *message_id = 0; + + if (connection->message.length + data_length > connection->buffer_length) + return fail_message(connection); + memcpy(connection->buffer + connection->message.length, data, data_length); + connection->message.length += data_length; + + return fini_message(connection, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_puback(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBACK, 0, 0, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pubrec(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBREC, 0, 0, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pubrel(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBREL, 0, 1, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pubcomp(mqtt_connection_t* connection, uint16_t message_id) +{ + init_message(connection); + if (append_message_id(connection, message_id) == 0) + return fail_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PUBCOMP, 0, 0, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_subscribe(mqtt_connection_t* connection, const char* topic, int qos, uint16_t* message_id) +{ + init_message(connection); + + if (topic == NULL || topic[0] == '\0') + return fail_message(connection); + + if ((*message_id = append_message_id(connection, 0)) == 0) + return fail_message(connection); + + if (append_string(connection, topic, strlen(topic)) < 0) + return fail_message(connection); + + if (connection->message.length + 1 > connection->buffer_length) + return fail_message(connection); + connection->buffer[connection->message.length++] = qos; + + return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* topic, uint16_t* message_id) +{ + init_message(connection); + + if (topic == NULL || topic[0] == '\0') + return fail_message(connection); + + if ((*message_id = append_message_id(connection, 0)) == 0) + return fail_message(connection); + + if (append_string(connection, topic, strlen(topic)) < 0) + return fail_message(connection); + + return fini_message(connection, MQTT_MSG_TYPE_UNSUBSCRIBE, 0, 1, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pingreq(mqtt_connection_t* connection) +{ + init_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_pingresp(mqtt_connection_t* connection) +{ + init_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_PINGRESP, 0, 0, 0); +} + +mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_disconnect(mqtt_connection_t* connection) +{ + init_message(connection); + return fini_message(connection, MQTT_MSG_TYPE_DISCONNECT, 0, 0, 0); +} diff --git a/mqtt/proto.c b/mqtt/proto.c new file mode 100644 index 0000000..3909b15 --- /dev/null +++ b/mqtt/proto.c @@ -0,0 +1,128 @@ +#include "proto.h" + +int8_t ICACHE_FLASH_ATTR PROTO_Init(PROTO_PARSER *parser, PROTO_PARSE_CALLBACK *completeCallback, uint8_t *buf, uint16_t bufSize) +{ + parser->buf = buf; + parser->bufSize = bufSize; + parser->dataLen = 0; + parser->callback = completeCallback; + parser->isEsc = 0; + return 0; +} + +int8_t ICACHE_FLASH_ATTR PROTO_ParseByte(PROTO_PARSER *parser, uint8_t value) +{ + switch(value){ + case 0x7D: + parser->isEsc = 1; + break; + + case 0x7E: + parser->dataLen = 0; + parser->isEsc = 0; + parser->isBegin = 1; + break; + + case 0x7F: + if (parser->callback != NULL) + parser->callback(); + parser->isBegin = 0; + return 0; + break; + + default: + if(parser->isBegin == 0) break; + + if(parser->isEsc){ + value ^= 0x20; + parser->isEsc = 0; + } + + if(parser->dataLen < parser->bufSize) + parser->buf[parser->dataLen++] = value; + + break; + } + return -1; +} + +int8_t ICACHE_FLASH_ATTR PROTO_Parse(PROTO_PARSER *parser, uint8_t *buf, uint16_t len) +{ + while(len--) + PROTO_ParseByte(parser, *buf++); + return 0; +} +int16_t ICACHE_FLASH_ATTR PROTO_ParseRb(RINGBUF* rb, uint8_t *bufOut, uint16_t* len, uint16_t maxBufLen) +{ + uint8_t c; + + PROTO_PARSER proto; + PROTO_Init(&proto, NULL, bufOut, maxBufLen); + while(RINGBUF_Get(rb, &c) == 0){ + if(PROTO_ParseByte(&proto, c) == 0){ + *len = proto.dataLen; + return 0; + } + } + return -1; +} +int16_t ICACHE_FLASH_ATTR PROTO_Add(uint8_t *buf, const uint8_t *packet, int16_t bufSize) +{ + uint16_t i = 2; + uint16_t len = *(uint16_t*) packet; + + if (bufSize < 1) return -1; + + *buf++ = 0x7E; + bufSize--; + + while (len--) { + switch (*packet) { + case 0x7D: + case 0x7E: + case 0x7F: + if (bufSize < 2) return -1; + *buf++ = 0x7D; + *buf++ = *packet++ ^ 0x20; + i += 2; + bufSize -= 2; + break; + default: + if (bufSize < 1) return -1; + *buf++ = *packet++; + i++; + bufSize--; + break; + } + } + + if (bufSize < 1) return -1; + *buf++ = 0x7F; + + return i; +} + +int16_t ICACHE_FLASH_ATTR PROTO_AddRb(RINGBUF *rb, const uint8_t *packet, int16_t len) +{ + uint16_t i = 2; + if(RINGBUF_Put(rb, 0x7E) == -1) return -1; + while (len--) { + switch (*packet) { + case 0x7D: + case 0x7E: + case 0x7F: + if(RINGBUF_Put(rb, 0x7D) == -1) return -1; + if(RINGBUF_Put(rb, *packet++ ^ 0x20) == -1) return -1; + i += 2; + break; + default: + if(RINGBUF_Put(rb, *packet++) == -1) return -1; + i++; + break; + } + } + if(RINGBUF_Put(rb, 0x7F) == -1) return -1; + + return i; +} + diff --git a/mqtt/queue.c b/mqtt/queue.c new file mode 100644 index 0000000..147015e --- /dev/null +++ b/mqtt/queue.c @@ -0,0 +1,50 @@ +/* str_queue.c +* +* Copyright (c) 2014-2015, Tuan PM +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions are met: +* +* * Redistributions of source code must retain the above copyright notice, +* this list of conditions and the following disclaimer. +* * Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* * Neither the name of Redis nor the names of its contributors may be used +* to endorse or promote products derived from this software without +* specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +*/ +#include "queue.h" + +void ICACHE_FLASH_ATTR QUEUE_Init(QUEUE *queue, int bufferSize) { + queue->buf = (uint8_t*)os_zalloc(bufferSize); + RINGBUF_Init(&queue->rb, queue->buf, bufferSize); +} + +int32_t ICACHE_FLASH_ATTR QUEUE_Puts(QUEUE *queue, uint8_t* buffer, uint16_t len) { + return PROTO_AddRb(&queue->rb, buffer, len); +} + +int32_t ICACHE_FLASH_ATTR QUEUE_Gets(QUEUE *queue, uint8_t* buffer, uint16_t* len, uint16_t maxLen) { + + return PROTO_ParseRb(&queue->rb, buffer, len, maxLen); +} + +bool ICACHE_FLASH_ATTR QUEUE_IsEmpty(QUEUE *queue) { + if (queue->rb.fill_cnt <= 0) + return TRUE; + return FALSE; +} \ No newline at end of file diff --git a/mqtt/ringbuf.c b/mqtt/ringbuf.c new file mode 100644 index 0000000..5ac3d07 --- /dev/null +++ b/mqtt/ringbuf.c @@ -0,0 +1,67 @@ +/** +* \file +* Ring Buffer library +*/ + +#include "ringbuf.h" + + +/** +* \brief init a RINGBUF object +* \param r pointer to a RINGBUF object +* \param buf pointer to a byte array +* \param size size of buf +* \return 0 if successfull, otherwise failed +*/ +I16 ICACHE_FLASH_ATTR RINGBUF_Init(RINGBUF *r, U8* buf, I32 size) +{ + if(r == NULL || buf == NULL || size < 2) return -1; + + r->p_o = r->p_r = r->p_w = buf; + r->fill_cnt = 0; + r->size = size; + + return 0; +} +/** +* \brief put a character into ring buffer +* \param r pointer to a ringbuf object +* \param c character to be put +* \return 0 if successfull, otherwise failed +*/ +I16 ICACHE_FLASH_ATTR RINGBUF_Put(RINGBUF *r, U8 c) +{ + if(r->fill_cnt>=r->size)return -1; // ring buffer is full, this should be atomic operation + + + r->fill_cnt++; // increase filled slots count, this should be atomic operation + + + *r->p_w++ = c; // put character into buffer + + if(r->p_w >= r->p_o + r->size) // rollback if write pointer go pass + r->p_w = r->p_o; // the physical boundary + + return 0; +} +/** +* \brief get a character from ring buffer +* \param r pointer to a ringbuf object +* \param c read character +* \return 0 if successfull, otherwise failed +*/ +I16 ICACHE_FLASH_ATTR RINGBUF_Get(RINGBUF *r, U8* c) +{ + if(r->fill_cnt<=0)return -1; // ring buffer is empty, this should be atomic operation + + + r->fill_cnt--; // decrease filled slots count + + + *c = *r->p_r++; // get the character out + + if(r->p_r >= r->p_o + r->size) // rollback if write pointer go pass + r->p_r = r->p_o; // the physical boundary + + return 0; +} diff --git a/mqtt/utils.c b/mqtt/utils.c new file mode 100644 index 0000000..5ece013 --- /dev/null +++ b/mqtt/utils.c @@ -0,0 +1,144 @@ +/* +* Copyright (c) 2014, Tuan PM +* Email: tuanpm@live.com +* +* All rights reserved. +* +* Redistribution and use in source and binary forms, with or without +* modification, are permitted provided that the following conditions +* are met: +* +* 1. Redistributions of source code must retain the above copyright +* notice, this list of conditions and the following disclaimer. +* 2. Redistributions in binary form must reproduce the above copyright +* notice, this list of conditions and the following disclaimer in the +* documentation and/or other materials provided with the distribution. +* 3. Neither the name of the copyright holder nor the names of its +* contributors may be used to endorse or promote products derived +* from this software without specific prior written permission. +* +* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +* POSSIBILITY OF SUCH DAMAGE. +* +*/ +#include "utils.h" + + +uint8_t ICACHE_FLASH_ATTR UTILS_IsIPV4 (int8_t *str) +{ + uint8_t segs = 0; /* Segment count. */ + uint8_t chcnt = 0; /* Character count within segment. */ + uint8_t accum = 0; /* Accumulator for segment. */ + /* Catch NULL pointer. */ + if (str == 0) + return 0; + /* Process every character in string. */ + + while (*str != '\0') { + /* Segment changeover. */ + + if (*str == '.') { + /* Must have some digits in segment. */ + if (chcnt == 0) + return 0; + /* Limit number of segments. */ + if (++segs == 4) + return 0; + /* Reset segment values and restart loop. */ + chcnt = accum = 0; + str++; + continue; + } + + /* Check numeric. */ + if ((*str < '0') || (*str > '9')) + return 0; + + /* Accumulate and check segment. */ + + if ((accum = accum * 10 + *str - '0') > 255) + return 0; + /* Advance other segment specific stuff and continue loop. */ + + chcnt++; + str++; + } + + /* Check enough segments and enough characters in last segment. */ + + if (segs != 3) + return 0; + if (chcnt == 0) + return 0; + /* Address okay. */ + + return 1; +} +uint8_t ICACHE_FLASH_ATTR UTILS_StrToIP(const int8_t* str, void *ip) +{ + + /* The count of the number of bytes processed. */ + int i; + /* A pointer to the next digit to process. */ + const char * start; + + start = str; + for (i = 0; i < 4; i++) { + /* The digit being processed. */ + char c; + /* The value of this byte. */ + int n = 0; + while (1) { + c = * start; + start++; + if (c >= '0' && c <= '9') { + n *= 10; + n += c - '0'; + } + /* We insist on stopping at "." if we are still parsing + the first, second, or third numbers. If we have reached + the end of the numbers, we will allow any character. */ + else if ((i < 3 && c == '.') || i == 3) { + break; + } + else { + return 0; + } + } + if (n >= 256) { + return 0; + } + ((uint8_t*)ip)[i] = n; + } + return 1; + +} +uint32_t ICACHE_FLASH_ATTR UTILS_Atoh(const int8_t *s) +{ + uint32_t value = 0, digit; + int8_t c; + + while((c = *s++)){ + if('0' <= c && c <= '9') + digit = c - '0'; + else if('A' <= c && c <= 'F') + digit = c - 'A' + 10; + else if('a' <= c && c<= 'f') + digit = c - 'a' + 10; + else break; + + value = (value << 4) | digit; + } + + return value; +} +