feedc0de/esp-mqtt
1
0
Fork 0
forked from espressif/esp-mqtt
Code Pull Requests Activity

Add mqtt connect msg

Browse Source
This commit is contained in:
Tuan PM
2016-09-11 21:42:34 +07:00
parent 89479c5734
commit eb1595f1e8
4 changed files with 189 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
Kconfig
View File

@@ -9,10 +9,15 @@ config MQTT_PROTOCOL_311
config MQTT_SECURITY_ON config MQTT_SECURITY_ON
bool "Enable MQTT over SSL" bool "Enable MQTT over SSL"
default y default n
help help
Enable MQTT Over SSL Enable MQTT Over SSL
config MQTT_PRIORITY
int "MQTT Task Priority"
range 1 15
default 5
config CONFIG_MQTT_LOG_ERROR_ON config CONFIG_MQTT_LOG_ERROR_ON
bool "Enable MQTT Debug message" bool "Enable MQTT Debug message"
default y default y
@@ -29,7 +34,7 @@ config CONFIG_MQTT_LOG_INFO_ON
help help
Disable it will redurce memory and run faster Disable it will redurce memory and run faster
config MQTT_RECONNECT_TIMEOUT config MQTT_RECONNECT_TIMEOUT
int "Reconnect timeout" int "Reconnect timeout (in second)"
range 10 16535 range 10 16535
default 60 default 60
@@ -38,6 +43,11 @@ config MQTT_QUEUE_BUFFER_SIZE_WORD
range 256 4096 range 256 4096
default 1024 default 1024
config MQTT_BUFFER_SIZE_BYTE
int "Network buffer size for MQTT in byte"
range 128 4096
default 1024
config MQTT_MAX_HOST_LEN config MQTT_MAX_HOST_LEN
int "Maximum host name len - in byte" int "Maximum host name len - in byte"
range 32 256 range 32 256

25
include/mqtt.h
View File

@@ -1,10 +1,12 @@
#ifndef _MQTT_H_ #ifndef _MQTT_H_
#define _MQTT_H_ #define _MQTT_H_
#include <stdint.h> #include <stdint.h>
#include <string.h>
#include "mqtt_config.h" #include "mqtt_config.h"
#include "mqtt_msg.h" #include "mqtt_msg.h"
typedef void (* mqtt_callback)(void *); typedef void (* mqtt_callback)(void *);
typedef struct { typedef struct {
mqtt_callback connected_cb; mqtt_callback connected_cb;
mqtt_callback disconnected_cb; mqtt_callback disconnected_cb;
@@ -21,12 +23,30 @@ typedef struct {
char password[CONFIG_MQTT_MAX_PASSWORD_LEN]; char password[CONFIG_MQTT_MAX_PASSWORD_LEN];
char lwt_topic[CONFIG_MQTT_MAX_LWT_TOPIC]; char lwt_topic[CONFIG_MQTT_MAX_LWT_TOPIC];
char lwt_msg[CONFIG_MQTT_MAX_LWT_MSG]; char lwt_msg[CONFIG_MQTT_MAX_LWT_MSG];
uint32_t lwt_qos, uint32_t lwt_qos;
uint32_t lwt_retain; uint32_t lwt_retain;
uint32_t clean_session; uint32_t clean_session;
uint32_t keepalive; uint32_t keepalive;
} mqtt_settings; } mqtt_settings;
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 struct { typedef struct {
int socket; int socket;
mqtt_settings *settings; mqtt_settings *settings;
@@ -34,7 +54,8 @@ typedef struct {
mqtt_connect_info_t connect_info; mqtt_connect_info_t connect_info;
} mqtt_client; } mqtt_client;
void mqtt_task(void *pv); void mqtt_start(mqtt_settings *mqtt_info);
void mqtt_task(void *pvParameters);
void mqtt_publish(); void mqtt_publish();
void mqtt_subscribe(); void mqtt_subscribe();
void mqtt_detroy(); void mqtt_detroy();

6
include/mqtt_config.h
View File

@@ -5,17 +5,17 @@
#ifdef CONFIG_MQTT_LOG_ERROR_ON #ifdef CONFIG_MQTT_LOG_ERROR_ON
#define mqtt_error( format, ... ) printf( "[ERROR] "format, ## __VA_ARGS__ ) #define mqtt_error( format, ... ) printf( "[MQTT ERROR] "format, ## __VA_ARGS__ )
#else #else
#define mqtt_error( format, ... ) #define mqtt_error( format, ... )
#endif #endif
#ifdef CONFIG_MQTT_LOG_WARN_ON #ifdef CONFIG_MQTT_LOG_WARN_ON
#define mqtt_warn( format, ... ) printf( "[WARN] "format, ## __VA_ARGS__ ) #define mqtt_warn( format, ... ) printf( "[MQTT WARN] "format, ## __VA_ARGS__ )
#else #else
#define mqtt_warn( format, ... ) #define mqtt_warn( format, ... )
#endif #endif
#ifdef CONFIG_MQTT_LOG_INFO_ON #ifdef CONFIG_MQTT_LOG_INFO_ON
#define mqtt_info( format, ... ) printf( "[INFO] "format, ## __VA_ARGS__ ) #define mqtt_info( format, ... ) printf( "[MQTT INFO] "format, ## __VA_ARGS__ )
#else #else
#define mqtt_info( format, ... ) #define mqtt_info( format, ... )
#endif #endif

153
mqtt.c
View File

@@ -2,12 +2,161 @@
* @Author: Tuan PM * @Author: Tuan PM
* @Date: 2016-09-10 09:33:06 * @Date: 2016-09-10 09:33:06
* @Last Modified by: Tuan PM * @Last Modified by: Tuan PM
* @Last Modified time: 2016-09-11 09:48:52 * @Last Modified time: 2016-09-11 21:41:55
*/ */
#include "mqtt.h" #include "mqtt.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/semphr.h"
#include "freertos/queue.h"
#include "lwip/sockets.h"
#include "lwip/dns.h"
#include "lwip/netdb.h"
static TaskHandle_t xMqttTask = NULL;
static int resolev_dns(const char *host, struct sockaddr_in *ip) {
struct hostent *he;
struct in_addr **addr_list;
he = gethostbyname(host);
if (he == NULL) return 0;
addr_list = (struct in_addr **)he->h_addr_list;
if (addr_list[0] == NULL) return 0;
ip->sin_family = AF_INET;
memcpy(&ip->sin_addr, addr_list[0], sizeof(ip->sin_addr));
return 1;
}
static int client_connect(const char *stream_host, int stream_port)
{
while (1) {
struct sockaddr_in remote_ip;
bzero(&remote_ip, sizeof(struct sockaddr_in));
//if stream_host is not ip address, resolve it
if (inet_pton(AF_INET, stream_host, &(remote_ip.sin_addr)) != 1) {
if (!resolev_dns(stream_host, &remote_ip)) {
vTaskDelay(1000 / portTICK_RATE_MS);
continue;
}
}
int sock = socket(PF_INET, SOCK_STREAM, 0);
if (sock == -1) {
continue;
}
remote_ip.sin_port = htons(stream_port);
mqtt_info("Connecting to server %s...:%d,%d\n", ipaddr_ntoa((const ip_addr_t*)&remote_ip.sin_addr.s_addr), stream_port, remote_ip.sin_port);
if (connect(sock, (struct sockaddr *)(&remote_ip), sizeof(struct sockaddr)) != 00) {
close(sock);
mqtt_error("[MQTT] Conn err.\n");
vTaskDelay(1000 / portTICK_RATE_MS);
continue;
}
return sock;
}
}
/*
* mqtt_connect
* input - client
* return 1: success, 0: fail
*/
static bool mqtt_connect(mqtt_client *client)
{
int write_len, read_len, connect_rsp_code;
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);
mqtt_info("Sending MQTT CONNECT message, type: %d, id: %04X\r\n",
client->mqtt_state.pending_msg_type,
client->mqtt_state.pending_msg_id);
write_len = write(client->socket,
client->mqtt_state.outbound_message->data,
client->mqtt_state.outbound_message->length);
mqtt_info("Reading MQTT CONNECT response message\n");
read_len = read(client->socket, client->mqtt_state.in_buffer, CONFIG_MQTT_BUFFER_SIZE_BYTE);
if (read_len == 0) {
mqtt_error("Error network response\n");
return false;
}
if (mqtt_get_type(client->mqtt_state.in_buffer) != MQTT_MSG_TYPE_CONNACK) {
mqtt_error("Invalid MSG_TYPE response: %d\n", mqtt_get_type(client->mqtt_state.in_buffer));
return false;
}
connect_rsp_code = mqtt_get_connect_return_code(client->mqtt_state.in_buffer);
switch (connect_rsp_code) {
case CONNECTION_ACCEPTED:
mqtt_info("Connected\n");
if (client->settings->connected_cb) {
client->settings->connected_cb(client);
}
return true;
case CONNECTION_REFUSE_PROTOCOL:
case CONNECTION_REFUSE_SERVER_UNAVAILABLE:
case CONNECTION_REFUSE_BAD_USERNAME:
case CONNECTION_REFUSE_NOT_AUTHORIZED:
mqtt_warn("Connection refuse, reason code: %d\r\n", connect_rsp_code);
return false;
default:
mqtt_warn("Connection refuse, Unknow reason\n");
return false;
}
return false;
}
void mqtt_task(void *pvParameters) void mqtt_task(void *pvParameters)
{ {
mqtt_settings *settings = (mqtt_settings *)pvParameters; mqtt_client *client = (mqtt_client *)pvParameters;
client->socket = client_connect(client->settings->host, client->settings->port);
mqtt_info("Connected to server %s:%d\n", client->host, client->port);
if (!mqtt_connect(client)) {
close(client->socket);
return;
} }
}
void mqtt_start(mqtt_settings *mqtt_info)
{
if (xMqttTask != NULL)
return;
mqtt_client *client = malloc(sizeof(mqtt_client));
if (client == NULL) {
mqtt_error("Memory not enought\n");
return;
}
client->settings = mqtt_info;
memset(&client->connect_info, 0, sizeof(mqtt_connect_info_t));
client->connect_info.client_id = mqtt_info->client_id;
client->connect_info.username = mqtt_info->username;
client->connect_info.password = mqtt_info->password;
client->connect_info.keepalive = mqtt_info->keepalive;
client->connect_info.clean_session = mqtt_info->clean_session;
client->mqtt_state.in_buffer = (uint8_t *)malloc(CONFIG_MQTT_BUFFER_SIZE_BYTE);
client->mqtt_state.in_buffer_length = CONFIG_MQTT_BUFFER_SIZE_BYTE;
client->mqtt_state.out_buffer = (uint8_t *)malloc(CONFIG_MQTT_BUFFER_SIZE_BYTE);
client->mqtt_state.out_buffer_length = CONFIG_MQTT_BUFFER_SIZE_BYTE;
client->mqtt_state.connect_info = &client->connect_info;
mqtt_msg_init(&client->mqtt_state.mqtt_connection,
client->mqtt_state.out_buffer,
client->mqtt_state.out_buffer_length);
xTaskCreate(&mqtt_task, "mqtt_task", 2048, &client, CONFIG_MQTT_PRIORITY, &xMqttTask);
}
void mqtt_stop()
{
}