mirror of
https://github.com/martin-ger/esp_wifi_repeater.git
synced 2026-07-27 19:56:03 +00:00
Initial integration of MQTT
This commit is contained in:
Martin Ger
@@ -33,10 +33,12 @@ ESPPORT ?= /dev/ttyUSB0
|
||||
TARGET = app
|
||||
|
||||
# which modules (subdirectories) of the project to include in compiling
|
||||
MODULES = driver user
|
||||
MODULES = driver user mqtt
|
||||
EXTRA_INCDIR = $(BUILD_AREA)/esp-open-sdk/esp-open-lwip/include include
|
||||
#EXTRA_INCDIR = include
|
||||
|
||||
LIB_MODULES = mqtt
|
||||
|
||||
# libraries used in this project, mainly provided by the SDK
|
||||
LIBS = c gcc hal pp phy net80211 lwip_open wpa main
|
||||
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
|
||||
#############################################################
|
||||
# Required variables for each makefile
|
||||
# Discard this section from all parent makefiles
|
||||
# Expected variables (with automatic defaults):
|
||||
# CSRCS (all "C" files in the dir)
|
||||
# SUBDIRS (all subdirs with a Makefile)
|
||||
# GEN_LIBS - list of libs to be generated ()
|
||||
# GEN_IMAGES - list of images to be generated ()
|
||||
# COMPONENTS_xxx - a list of libs/objs in the form
|
||||
# subdir/lib to be extracted and rolled up into
|
||||
# a generated lib/image xxx.a ()
|
||||
#
|
||||
ifndef PDIR
|
||||
GEN_LIBS = libmqtt.a
|
||||
endif
|
||||
|
||||
|
||||
#############################################################
|
||||
# Configuration i.e. compile options etc.
|
||||
# Target specific stuff (defines etc.) goes in here!
|
||||
# Generally values applying to a tree are captured in the
|
||||
# makefile at its root level - these are then overridden
|
||||
# for a subtree within the makefile rooted therein
|
||||
#
|
||||
#DEFINES +=
|
||||
|
||||
#############################################################
|
||||
# Recursion Magic - Don't touch this!!
|
||||
#
|
||||
# Each subtree potentially has an include directory
|
||||
# corresponding to the common APIs applicable to modules
|
||||
# rooted at that subtree. Accordingly, the INCLUDE PATH
|
||||
# of a module can only contain the include directories up
|
||||
# its parent path, and not its siblings
|
||||
#
|
||||
# Required for each makefile to inherit from the parent
|
||||
#
|
||||
|
||||
INCLUDES := $(INCLUDES) -I $(PDIR)include
|
||||
INCLUDES += -I ./
|
||||
PDIR := ../$(PDIR)
|
||||
sinclude $(PDIR)Makefile
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
/*
|
||||
* debug.h
|
||||
*
|
||||
* Created on: Dec 4, 2014
|
||||
* Author: Minh
|
||||
*/
|
||||
|
||||
#ifndef USER_DEBUG_H_
|
||||
#define USER_DEBUG_H_
|
||||
|
||||
#if defined(MQTT_DEBUG_ON)
|
||||
#define MQTT_INFO( format, ... ) os_printf( format, ## __VA_ARGS__ )
|
||||
#else
|
||||
#define MQTT_INFO( format, ... )
|
||||
#endif
|
||||
|
||||
|
||||
#endif /* USER_DEBUG_H_ */
|
||||
@@ -0,0 +1,148 @@
|
||||
/* mqtt.h
|
||||
*
|
||||
* Copyright (c) 2014-2015, Tuan PM <tuanpm at live dot 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:
|
||||
*
|
||||
* * 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 "user_config.h"
|
||||
#include "mqtt_msg.h"
|
||||
#include "user_interface.h"
|
||||
|
||||
#include "queue.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_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;
|
||||
|
||||
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;
|
||||
MqttCallback timeoutCb;
|
||||
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_t port, uint8_t security);
|
||||
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);
|
||||
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);
|
||||
BOOL ICACHE_FLASH_ATTR MQTT_UnSubscribe(MQTT_Client *client, char* topic);
|
||||
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 data_length, int qos, int retain);
|
||||
|
||||
#endif /* USER_AT_MQTT_H_ */
|
||||
@@ -0,0 +1,141 @@
|
||||
/*
|
||||
* File: mqtt_msg.h
|
||||
* Author: Minh Tuan
|
||||
*
|
||||
* Created on July 12, 2014, 1:05 PM
|
||||
*/
|
||||
|
||||
#ifndef MQTT_MSG_H
|
||||
#define MQTT_MSG_H
|
||||
#include "user_config.h"
|
||||
#include "c_types.h"
|
||||
#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
|
||||
};
|
||||
|
||||
enum mqtt_connect_return_code
|
||||
{
|
||||
CONNECTION_ACCEPTED = 0,
|
||||
CONNECTION_REFUSE_PROTOCOL,
|
||||
CONNECTION_REFUSE_ID_REJECTED,
|
||||
CONNECTION_REFUSE_SERVER_UNAVAILABLE,
|
||||
CONNECTION_REFUSE_BAD_USERNAME,
|
||||
CONNECTION_REFUSE_NOT_AUTHORIZED
|
||||
};
|
||||
|
||||
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;
|
||||
uint32_t 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_connect_return_code(uint8_t* buffer) { return buffer[3]; }
|
||||
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 */
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
/*
|
||||
* File: proto.h
|
||||
* Author: ThuHien
|
||||
*
|
||||
* Created on November 23, 2012, 8:57 AM
|
||||
*/
|
||||
|
||||
#ifndef _PROTO_H_
|
||||
#define _PROTO_H_
|
||||
#include <stdlib.h>
|
||||
#include "typedef.h"
|
||||
#include "ringbuf_mqtt.h"
|
||||
|
||||
typedef void(PROTO_PARSE_CALLBACK)();
|
||||
|
||||
typedef struct {
|
||||
U8 *buf;
|
||||
U16 bufSize;
|
||||
U16 dataLen;
|
||||
U8 isEsc;
|
||||
U8 isBegin;
|
||||
PROTO_PARSE_CALLBACK* callback;
|
||||
} PROTO_PARSER;
|
||||
|
||||
I8 ICACHE_FLASH_ATTR PROTO_Init(PROTO_PARSER *parser, PROTO_PARSE_CALLBACK *completeCallback, U8 *buf, U16 bufSize);
|
||||
I8 ICACHE_FLASH_ATTR PROTO_Parse(PROTO_PARSER *parser, U8 *buf, U16 len);
|
||||
I16 ICACHE_FLASH_ATTR PROTO_Add(U8 *buf, const U8 *packet, I16 bufSize);
|
||||
I16 ICACHE_FLASH_ATTR PROTO_AddRb(RINGBUF *rb, const U8 *packet, I16 len);
|
||||
I8 ICACHE_FLASH_ATTR PROTO_ParseByte(PROTO_PARSER *parser, U8 value);
|
||||
I16 ICACHE_FLASH_ATTR PROTO_ParseRb(RINGBUF *rb, U8 *bufOut, U16* len, U16 maxBufLen);
|
||||
#endif
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
/* str_queue.h --
|
||||
*
|
||||
* Copyright (c) 2014-2015, Tuan PM <tuanpm at live dot 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:
|
||||
*
|
||||
* * 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 "os_type.h"
|
||||
#include "ringbuf_mqtt.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_ */
|
||||
@@ -0,0 +1,19 @@
|
||||
#ifndef _RING_BUF_MQTT_H_
|
||||
#define _RING_BUF_MQTT_H_
|
||||
|
||||
#include <os_type.h>
|
||||
#include <stdlib.h>
|
||||
#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
|
||||
@@ -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
|
||||
@@ -0,0 +1,9 @@
|
||||
#ifndef _UTILS_H_
|
||||
#define _UTILS_H_
|
||||
|
||||
#include "c_types.h"
|
||||
|
||||
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
|
||||
+995
@@ -0,0 +1,995 @@
|
||||
/* 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 <tuanpm at live dot 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:
|
||||
*
|
||||
* * 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 "user_interface.h"
|
||||
#include "osapi.h"
|
||||
#include "espconn.h"
|
||||
#include "os_type.h"
|
||||
#include "mem.h"
|
||||
#include "mqtt_msg.h"
|
||||
#include "debug.h"
|
||||
#include "user_config.h"
|
||||
#include "mqtt.h"
|
||||
#include "queue.h"
|
||||
|
||||
#define MQTT_TASK_PRIO 2
|
||||
#define MQTT_TASK_QUEUE_SIZE 1
|
||||
#define MQTT_SEND_TIMOUT 5
|
||||
|
||||
#ifndef MQTT_SSL_SIZE
|
||||
#define MQTT_SSL_SIZE 5120
|
||||
#endif
|
||||
|
||||
#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];
|
||||
|
||||
#ifdef PROTOCOL_NAMEv311
|
||||
LOCAL uint8_t zero_len_id[2] = { 0, 0 };
|
||||
#endif
|
||||
|
||||
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)
|
||||
{
|
||||
MQTT_INFO("DNS: Found, but got no ip, try to reconnect\r\n");
|
||||
client->connState = TCP_RECONNECT_REQ;
|
||||
return;
|
||||
}
|
||||
|
||||
MQTT_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_set_size(ESPCONN_CLIENT, MQTT_SSL_SIZE);
|
||||
espconn_secure_connect(client->pCon);
|
||||
#else
|
||||
MQTT_INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
espconn_connect(client->pCon);
|
||||
}
|
||||
|
||||
client->connState = TCP_CONNECTING;
|
||||
MQTT_INFO("TCP: connecting...\r\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);
|
||||
|
||||
}
|
||||
|
||||
void ICACHE_FLASH_ATTR
|
||||
mqtt_send_keepalive(MQTT_Client *client)
|
||||
{
|
||||
MQTT_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;
|
||||
MQTT_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
|
||||
MQTT_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);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Delete tcp client and free all memory
|
||||
* @param mqttClient: The mqtt client which contain TCP client
|
||||
* @retval None
|
||||
*/
|
||||
void ICACHE_FLASH_ATTR
|
||||
mqtt_tcpclient_delete(MQTT_Client *mqttClient)
|
||||
{
|
||||
if (mqttClient->pCon != NULL) {
|
||||
MQTT_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;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Delete MQTT client and free all memory
|
||||
* @param mqttClient: The mqtt client
|
||||
* @retval None
|
||||
*/
|
||||
void ICACHE_FLASH_ATTR
|
||||
mqtt_client_delete(MQTT_Client *mqttClient)
|
||||
{
|
||||
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) {
|
||||
#ifdef PROTOCOL_NAMEv311
|
||||
/* Don't attempt to free if it's the zero_len array */
|
||||
if ( ((uint8_t*)mqttClient->connect_info.client_id) != zero_len_id )
|
||||
os_free(mqttClient->connect_info.client_id);
|
||||
#else
|
||||
os_free(mqttClient->connect_info.client_id);
|
||||
#endif
|
||||
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;
|
||||
|
||||
MQTT_INFO("MQTT: client already deleted\r\n");
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @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;
|
||||
uint8_t msg_conn_ret;
|
||||
|
||||
struct espconn *pCon = (struct espconn*)arg;
|
||||
MQTT_Client *client = (MQTT_Client *)pCon->reverse;
|
||||
|
||||
client->keepAliveTick = 0;
|
||||
READPACKET:
|
||||
MQTT_INFO("TCP: data received %d bytes\r\n", len);
|
||||
// MQTT_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) {
|
||||
MQTT_INFO("MQTT: Invalid packet\r\n");
|
||||
if (client->security) {
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_disconnect(client->pCon);
|
||||
#else
|
||||
MQTT_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:
|
||||
MQTT_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:
|
||||
MQTT_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
|
||||
MQTT_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)
|
||||
MQTT_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)
|
||||
MQTT_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) {
|
||||
MQTT_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) {
|
||||
MQTT_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) {
|
||||
MQTT_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) {
|
||||
MQTT_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) {
|
||||
MQTT_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) {
|
||||
MQTT_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) {
|
||||
MQTT_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;
|
||||
|
||||
MQTT_INFO("Get another published message\r\n");
|
||||
goto READPACKET;
|
||||
}
|
||||
|
||||
}
|
||||
break;
|
||||
}
|
||||
} else {
|
||||
MQTT_INFO("ERROR: Message too long\r\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;
|
||||
MQTT_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);
|
||||
}
|
||||
|
||||
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 / 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 --;
|
||||
}
|
||||
|
||||
void ICACHE_FLASH_ATTR
|
||||
mqtt_tcpclient_discon_cb(void *arg)
|
||||
{
|
||||
|
||||
struct espconn *pespconn = (struct espconn *)arg;
|
||||
MQTT_Client* client = (MQTT_Client *)pespconn->reverse;
|
||||
MQTT_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);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* @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);///////
|
||||
MQTT_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;
|
||||
MQTT_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
|
||||
MQTT_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);
|
||||
}
|
||||
|
||||
/**
|
||||
* @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;
|
||||
|
||||
MQTT_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);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @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 data_length, int qos, int retain)
|
||||
{
|
||||
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) {
|
||||
MQTT_INFO("MQTT: Queuing publish failed\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
MQTT_INFO("MQTT: queuing publish, length: %d, queue size(%d/%d)\r\n", client->mqtt_state.outbound_message->length, client->msgQueue.rb.fill_cnt, client->msgQueue.rb.size);
|
||||
while (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
MQTT_INFO("MQTT: Queue full\r\n");
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
MQTT_INFO("MQTT: Serious buffer error\r\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, qos,
|
||||
&client->mqtt_state.pending_msg_id);
|
||||
MQTT_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) {
|
||||
MQTT_INFO("MQTT: Queue full\r\n");
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
MQTT_INFO("MQTT: Serious buffer error\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
}
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief MQTT un-subscibe function.
|
||||
* @param client: MQTT_Client reference
|
||||
* @param topic: String topic will un-subscribe
|
||||
* @retval TRUE if success queue
|
||||
*/
|
||||
BOOL ICACHE_FLASH_ATTR
|
||||
MQTT_UnSubscribe(MQTT_Client *client, char* topic)
|
||||
{
|
||||
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);
|
||||
MQTT_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) {
|
||||
MQTT_INFO("MQTT: Queue full\r\n");
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
MQTT_INFO("MQTT: Serious buffer error\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
}
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client);
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief MQTT ping function.
|
||||
* @param client: MQTT_Client reference
|
||||
* @retval TRUE if success queue
|
||||
*/
|
||||
BOOL ICACHE_FLASH_ATTR
|
||||
MQTT_Ping(MQTT_Client *client)
|
||||
{
|
||||
uint8_t dataBuffer[MQTT_BUF_SIZE];
|
||||
uint16_t dataLen;
|
||||
client->mqtt_state.outbound_message = mqtt_msg_pingreq(&client->mqtt_state.mqtt_connection);
|
||||
if (client->mqtt_state.outbound_message->length == 0) {
|
||||
MQTT_INFO("MQTT: Queuing publish failed\r\n");
|
||||
return FALSE;
|
||||
}
|
||||
MQTT_INFO("MQTT: queuing publish, length: %d, queue size(%d/%d)\r\n", client->mqtt_state.outbound_message->length, client->msgQueue.rb.fill_cnt, client->msgQueue.rb.size);
|
||||
while (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1) {
|
||||
MQTT_INFO("MQTT: Queue full\r\n");
|
||||
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) {
|
||||
MQTT_INFO("MQTT: Serious buffer error\r\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;
|
||||
switch (client->connState) {
|
||||
|
||||
case TCP_RECONNECT_REQ:
|
||||
break;
|
||||
case TCP_RECONNECT:
|
||||
mqtt_tcpclient_delete(client);
|
||||
MQTT_Connect(client);
|
||||
MQTT_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
|
||||
MQTT_INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
espconn_disconnect(client->pCon);
|
||||
}
|
||||
break;
|
||||
case TCP_DISCONNECTED:
|
||||
MQTT_INFO("MQTT: Disconnected\r\n");
|
||||
mqtt_tcpclient_delete(client);
|
||||
break;
|
||||
case MQTT_DELETED:
|
||||
MQTT_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;
|
||||
MQTT_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
|
||||
MQTT_INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
espconn_send(client->pCon, dataBuffer, dataLen);
|
||||
}
|
||||
|
||||
client->mqtt_state.outbound_message = NULL;
|
||||
break;
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @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_t port, uint8_t security)
|
||||
{
|
||||
uint32_t temp;
|
||||
MQTT_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;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @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
|
||||
*/
|
||||
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)
|
||||
{
|
||||
uint32_t temp;
|
||||
MQTT_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 {
|
||||
MQTT_INFO("cleanSession must be set to use 0 length client_id\r\n");
|
||||
return false;
|
||||
}
|
||||
/* Not supported. Return. */
|
||||
#else
|
||||
MQTT_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;
|
||||
}
|
||||
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(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;
|
||||
}
|
||||
/**
|
||||
* @brief Begin connect to MQTT broker
|
||||
* @param client: MQTT_Client reference
|
||||
* @retval None
|
||||
*/
|
||||
void ICACHE_FLASH_ATTR
|
||||
MQTT_Connect(MQTT_Client *mqttClient)
|
||||
{
|
||||
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;
|
||||
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)) {
|
||||
MQTT_INFO("TCP: Connect to ip %s:%d\r\n", mqttClient->host, mqttClient->port);
|
||||
if (mqttClient->security)
|
||||
{
|
||||
#ifdef MQTT_SSL_ENABLE
|
||||
espconn_secure_set_size(ESPCONN_CLIENT, MQTT_SSL_SIZE);
|
||||
espconn_secure_connect(mqttClient->pCon);
|
||||
#else
|
||||
MQTT_INFO("TCP: Do not support SSL\r\n");
|
||||
#endif
|
||||
}
|
||||
else
|
||||
{
|
||||
espconn_connect(mqttClient->pCon);
|
||||
}
|
||||
}
|
||||
else {
|
||||
MQTT_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;
|
||||
}
|
||||
|
||||
void ICACHE_FLASH_ATTR
|
||||
MQTT_Disconnect(MQTT_Client *mqttClient)
|
||||
{
|
||||
mqttClient->connState = TCP_DISCONNECTING;
|
||||
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient);
|
||||
os_timer_disarm(&mqttClient->mqttTimer);
|
||||
}
|
||||
|
||||
void ICACHE_FLASH_ATTR
|
||||
MQTT_DeleteClient(MQTT_Client *mqttClient)
|
||||
{
|
||||
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);
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
void ICACHE_FLASH_ATTR
|
||||
MQTT_OnTimeout(MQTT_Client *mqttClient, MqttCallback timeoutCb)
|
||||
{
|
||||
mqttClient->timeoutCb = timeoutCb;
|
||||
}
|
||||
+493
@@ -0,0 +1,493 @@
|
||||
/*
|
||||
* 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 "mem.h"
|
||||
#include "ets_sys.h"
|
||||
#include "osapi.h"
|
||||
#include "os_type.h"
|
||||
|
||||
#include <string.h>
|
||||
#include "mqtt_msg.h"
|
||||
#include "user_config.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;
|
||||
os_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)
|
||||
{
|
||||
os_memset(connection, 0, sizeof(mqtt_connection_t));
|
||||
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;
|
||||
int blength = *length;
|
||||
*length = 0;
|
||||
|
||||
for (i = 1; i < blength; ++i)
|
||||
{
|
||||
totlen += (buffer[i] & 0x7f) << (7 * (i - 1));
|
||||
if ((buffer[i] & 0x80) == 0)
|
||||
{
|
||||
++i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
totlen += i;
|
||||
|
||||
if (i + 2 >= blength)
|
||||
return NULL;
|
||||
topiclen = buffer[i++] << 8;
|
||||
topiclen |= buffer[i++];
|
||||
|
||||
if (i + topiclen >= blength)
|
||||
return NULL;
|
||||
|
||||
i += topiclen;
|
||||
|
||||
if (mqtt_get_qos(buffer) > 0)
|
||||
{
|
||||
if (i + 2 >= blength)
|
||||
return NULL;
|
||||
i += 2;
|
||||
}
|
||||
|
||||
if (totlen < i)
|
||||
return NULL;
|
||||
|
||||
if (totlen <= blength)
|
||||
*length = totlen - i;
|
||||
else
|
||||
*length = blength - 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;
|
||||
os_memcpy(variable_header->magic, "MQIsdp", 6);
|
||||
variable_header->version = 3;
|
||||
#elif defined(PROTOCOL_NAMEv311)
|
||||
variable_header->lengthLsb = 4;
|
||||
os_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)
|
||||
{
|
||||
/* Never allowed */
|
||||
return fail_message(connection);
|
||||
}
|
||||
else if (info->client_id[0] == '\0')
|
||||
{
|
||||
#ifdef PROTOCOL_NAMEv311
|
||||
/* Allowed. Format 0 Length ID */
|
||||
append_string(connection, info->client_id, 2) ;
|
||||
#else
|
||||
/* 0 Length not allowed */
|
||||
return fail_message(connection);
|
||||
#endif
|
||||
}
|
||||
else
|
||||
{
|
||||
/* No 0 data and at least 1 long. Good to go. */
|
||||
if(append_string(connection, info->client_id, os_strlen(info->client_id)) < 0)
|
||||
return fail_message(connection);
|
||||
}
|
||||
|
||||
if (info->will_topic != NULL && info->will_topic[0] != '\0')
|
||||
{
|
||||
if (append_string(connection, info->will_topic, os_strlen(info->will_topic)) < 0)
|
||||
return fail_message(connection);
|
||||
|
||||
if (append_string(connection, info->will_message, os_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, os_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, os_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, os_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);
|
||||
os_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, os_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, os_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);
|
||||
}
|
||||
+129
@@ -0,0 +1,129 @@
|
||||
#include "proto.h"
|
||||
#include "ringbuf_mqtt.h"
|
||||
I8 ICACHE_FLASH_ATTR PROTO_Init(PROTO_PARSER *parser, PROTO_PARSE_CALLBACK *completeCallback, U8 *buf, U16 bufSize)
|
||||
{
|
||||
parser->buf = buf;
|
||||
parser->bufSize = bufSize;
|
||||
parser->dataLen = 0;
|
||||
parser->callback = completeCallback;
|
||||
parser->isEsc = 0;
|
||||
return 0;
|
||||
}
|
||||
|
||||
I8 ICACHE_FLASH_ATTR PROTO_ParseByte(PROTO_PARSER *parser, U8 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;
|
||||
}
|
||||
|
||||
I8 ICACHE_FLASH_ATTR PROTO_Parse(PROTO_PARSER *parser, U8 *buf, U16 len)
|
||||
{
|
||||
while (len--)
|
||||
PROTO_ParseByte(parser, *buf++);
|
||||
|
||||
return 0;
|
||||
}
|
||||
I16 ICACHE_FLASH_ATTR PROTO_ParseRb(RINGBUF* rb, U8 *bufOut, U16* len, U16 maxBufLen)
|
||||
{
|
||||
U8 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;
|
||||
}
|
||||
I16 ICACHE_FLASH_ATTR PROTO_Add(U8 *buf, const U8 *packet, I16 bufSize)
|
||||
{
|
||||
U16 i = 2;
|
||||
U16 len = *(U16*) 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;
|
||||
}
|
||||
|
||||
I16 ICACHE_FLASH_ATTR PROTO_AddRb(RINGBUF *rb, const U8 *packet, I16 len)
|
||||
{
|
||||
U16 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;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
/* str_queue.c
|
||||
*
|
||||
* Copyright (c) 2014-2015, Tuan PM <tuanpm at live dot 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:
|
||||
*
|
||||
* * 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"
|
||||
|
||||
#include "user_interface.h"
|
||||
#include "osapi.h"
|
||||
#include "os_type.h"
|
||||
#include "mem.h"
|
||||
#include "proto.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;
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
/**
|
||||
* \file
|
||||
* Ring Buffer library
|
||||
*/
|
||||
|
||||
#include "ringbuf_mqtt.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;
|
||||
}
|
||||
+149
@@ -0,0 +1,149 @@
|
||||
/*
|
||||
* 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 <string.h>
|
||||
#include <stdio.h>
|
||||
#include <ctype.h>
|
||||
#include <math.h>
|
||||
#include <stddef.h>
|
||||
#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;
|
||||
}
|
||||
|
||||
@@ -62,4 +62,16 @@ typedef enum {SIG_DO_NOTHING=0, SIG_START_SERVER=1, SIG_SEND_DATA, SIG_UART0, SI
|
||||
// Define this if you want to silently drop any packet that cannot be send to the monitor
|
||||
//#define DROP_PACKET_IF_NOT_RECORDED 1
|
||||
|
||||
//
|
||||
// Here the MQTT stuff
|
||||
//
|
||||
// Define this if you want to have it work as a MQTT client
|
||||
#define MQTT_CLIENT 1
|
||||
|
||||
#define MQTT_BUF_SIZE 1024
|
||||
#define MQTT_KEEPALIVE 120 /*second*/
|
||||
#define MQTT_RECONNECT_TIMEOUT 5 /*second*/
|
||||
#define PROTOCOL_NAMEv31 /*MQTT version 3.1 compatible with Mosquitto v0.15*/
|
||||
//#define PROTOCOL_NAMEv311 /*MQTT version 3.11 compatible with https://eclipse.org/paho/clients/testing/*/
|
||||
|
||||
#endif
|
||||
|
||||
@@ -57,6 +57,62 @@ uint8_t remote_console_disconnect;
|
||||
void ICACHE_FLASH_ATTR user_set_softap_wifi_config(void);
|
||||
void ICACHE_FLASH_ATTR user_set_softap_ip_config(void);
|
||||
|
||||
#ifdef MQTT_CLIENT
|
||||
#include "mqtt.h"
|
||||
MQTT_Client mqttClient;
|
||||
|
||||
static void ICACHE_FLASH_ATTR mqttConnectedCb(uint32_t *args)
|
||||
{
|
||||
MQTT_Client* client = (MQTT_Client*)args;
|
||||
os_printf("MQTT: Connected\r\n");
|
||||
MQTT_Subscribe(client, "/mqtt/WiFiRepeater/command", 0);
|
||||
MQTT_Subscribe(client, "/mqtt/topic/1", 1);
|
||||
MQTT_Subscribe(client, "/mqtt/topic/2", 2);
|
||||
|
||||
MQTT_Publish(client, "/mqtt/topic/0", "hello0", 6, 0, 0);
|
||||
MQTT_Publish(client, "/mqtt/topic/1", "hello1", 6, 1, 0);
|
||||
MQTT_Publish(client, "/mqtt/topic/2", "hello2", 6, 2, 0);
|
||||
|
||||
}
|
||||
|
||||
static void ICACHE_FLASH_ATTR mqttDisconnectedCb(uint32_t *args)
|
||||
{
|
||||
MQTT_Client* client = (MQTT_Client*)args;
|
||||
os_printf("MQTT: Disconnected\r\n");
|
||||
}
|
||||
|
||||
static void ICACHE_FLASH_ATTR mqttPublishedCb(uint32_t *args)
|
||||
{
|
||||
MQTT_Client* client = (MQTT_Client*)args;
|
||||
os_printf("MQTT: Published\r\n");
|
||||
}
|
||||
|
||||
static void ICACHE_FLASH_ATTR mqttDataCb(uint32_t *args, const char* topic, uint32_t topic_len, const char *data, uint32_t data_len)
|
||||
{
|
||||
char *topicBuf = (char*)os_zalloc(topic_len + 1),
|
||||
*dataBuf = (char*)os_zalloc(data_len + 1);
|
||||
|
||||
MQTT_Client* client = (MQTT_Client*)args;
|
||||
|
||||
if (os_strncmp(topic,"/mqtt/WiFiRepeater/command", topic_len) == 0) {
|
||||
ringbuf_memcpy_into(console_rx_buffer, data, data_len);
|
||||
ringbuf_memcpy_into(console_rx_buffer, "\n", 1);
|
||||
// signal the main task that command is available for processing
|
||||
system_os_post(0, SIG_CONSOLE_RX, 0);
|
||||
return;
|
||||
}
|
||||
|
||||
os_memcpy(topicBuf, topic, topic_len);
|
||||
topicBuf[topic_len] = 0;
|
||||
os_memcpy(dataBuf, data, data_len);
|
||||
dataBuf[data_len] = 0;
|
||||
os_printf("Receive topic: %s, data: %s \r\n", topicBuf, dataBuf);
|
||||
os_free(topicBuf);
|
||||
os_free(dataBuf);
|
||||
}
|
||||
#endif /* MQTT_CLIENT */
|
||||
|
||||
|
||||
#ifdef REMOTE_MONITORING
|
||||
static uint8_t monitoring_on;
|
||||
static uint16_t monitor_port;
|
||||
@@ -990,6 +1046,11 @@ void wifi_handle_event_cb(System_Event_t *evt)
|
||||
case EVENT_STAMODE_DISCONNECTED:
|
||||
os_printf("disconnect from ssid %s, reason %d\n", evt->event_info.disconnected.ssid, evt->event_info.disconnected.reason);
|
||||
connected = false;
|
||||
|
||||
#ifdef MQTT_CLIENT
|
||||
MQTT_Disconnect(&mqttClient);
|
||||
#endif /* MQTT_CLIENT */
|
||||
|
||||
break;
|
||||
|
||||
case EVENT_STAMODE_AUTHMODE_CHANGE:
|
||||
@@ -1012,6 +1073,10 @@ void wifi_handle_event_cb(System_Event_t *evt)
|
||||
}
|
||||
}
|
||||
|
||||
#ifdef MQTT_CLIENT
|
||||
MQTT_Connect(&mqttClient);
|
||||
#endif /* MQTT_CLIENT */
|
||||
|
||||
// Post a Server Start message as the IP has been acquired to Task with priority 0
|
||||
system_os_post(user_procTaskPrio, SIG_START_SERVER, 0 );
|
||||
break;
|
||||
@@ -1191,6 +1256,22 @@ void ICACHE_FLASH_ATTR user_init()
|
||||
monitor_port = 0;
|
||||
#endif
|
||||
|
||||
#ifdef MQTT_CLIENT
|
||||
//MQTT_InitConnection(&mqttClient, MQTT_HOST, MQTT_PORT, DEFAULT_SECURITY);
|
||||
MQTT_InitConnection(&mqttClient, "192.168.178.39", 1883, 0);
|
||||
|
||||
// if ( !MQTT_InitClient(&mqttClient, MQTT_CLIENT_ID, MQTT_USER, MQTT_PASS, MQTT_KEEPALIVE, MQTT_CLEAN_SESSION) )
|
||||
if (!MQTT_InitClient(&mqttClient, "WiFiRepeater", 0, 0, 120, 1))
|
||||
{
|
||||
os_printf("Failed to initialize properly.\r\n");
|
||||
}
|
||||
// MQTT_InitLWT(&mqttClient, "/lwt", "offline", 0, 0);
|
||||
MQTT_OnConnected(&mqttClient, mqttConnectedCb);
|
||||
MQTT_OnDisconnected(&mqttClient, mqttDisconnectedCb);
|
||||
MQTT_OnPublished(&mqttClient, mqttPublishedCb);
|
||||
MQTT_OnData(&mqttClient, mqttDataCb);
|
||||
#endif /* MQTT_CLIENT */
|
||||
|
||||
remote_console_disconnect = 0;
|
||||
|
||||
// Now start the STA-Mode
|
||||
|
||||
Reference in New Issue
Block a user