Commit 09750b56 authored by funshine's avatar funshine
Browse files

add queue to mqtt module

parent 3dd1ac37
...@@ -46,7 +46,7 @@ int strbuf_init(strbuf_t *s, int len) ...@@ -46,7 +46,7 @@ int strbuf_init(strbuf_t *s, int len)
s->reallocs = 0; s->reallocs = 0;
s->debug = 0; s->debug = 0;
s->buf = c_malloc(size); s->buf = (char *)c_malloc(size);
if (!s->buf){ if (!s->buf){
NODE_ERR("not enough memory\n"); NODE_ERR("not enough memory\n");
return -1; return -1;
...@@ -60,7 +60,7 @@ strbuf_t *strbuf_new(int len) ...@@ -60,7 +60,7 @@ strbuf_t *strbuf_new(int len)
{ {
strbuf_t *s; strbuf_t *s;
s = c_malloc(sizeof(strbuf_t)); s = (strbuf_t *)c_malloc(sizeof(strbuf_t));
if (!s){ if (!s){
NODE_ERR("not enough memory\n"); NODE_ERR("not enough memory\n");
return NULL; return NULL;
......
...@@ -7,6 +7,6 @@ ...@@ -7,6 +7,6 @@
#define NODE_VERSION_INTERNAL 0U #define NODE_VERSION_INTERNAL 0U
#define NODE_VERSION "NodeMCU 0.9.5" #define NODE_VERSION "NodeMCU 0.9.5"
#define BUILD_DATE "build 20150318" #define BUILD_DATE "build 20150330"
#endif /* __USER_VERSION_H__ */ #endif /* __USER_VERSION_H__ */
...@@ -47,9 +47,9 @@ extern int c_stderr; ...@@ -47,9 +47,9 @@ extern int c_stderr;
#define SEEK_END 2 /* set file offset to EOF plus offset */ #define SEEK_END 2 /* set file offset to EOF plus offset */
#endif #endif
#define c_malloc os_malloc // #define c_malloc os_malloc
#define c_zalloc os_zalloc // #define c_zalloc os_zalloc
#define c_free os_free // #define c_free os_free
extern void output_redirect(const char *str); extern void output_redirect(const char *str);
#define c_puts output_redirect #define c_puts output_redirect
......
...@@ -29,9 +29,9 @@ ...@@ -29,9 +29,9 @@
#define os_realloc(p, s) mem_realloc((p), (s)) #define os_realloc(p, s) mem_realloc((p), (s))
#endif #endif
// #define c_free os_free #define c_free os_free
// #define c_malloc os_malloc #define c_malloc os_malloc
// #define c_zalloc os_zalloc #define c_zalloc os_zalloc
#define c_realloc os_realloc #define c_realloc os_realloc
#define c_abs abs #define c_abs abs
...@@ -47,9 +47,9 @@ ...@@ -47,9 +47,9 @@
// c_getenv() get env "LUA_INIT" string for lua initialization. // c_getenv() get env "LUA_INIT" string for lua initialization.
const char *c_getenv(const char *__string); const char *c_getenv(const char *__string);
void *c_malloc(size_t __size); // void *c_malloc(size_t __size);
void *c_zalloc(size_t __size); // void *c_zalloc(size_t __size);
void c_free(void *); // void c_free(void *);
// int c_rand(void); // int c_rand(void);
// void c_srand(unsigned int __seed); // void c_srand(unsigned int __seed);
......
This diff is collapsed.
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
* *
*/ */
#include <string.h> #include "c_string.h"
#include "mqtt_msg.h" #include "mqtt_msg.h"
#define MQTT_MAX_FIXED_HEADER_SIZE 3 #define MQTT_MAX_FIXED_HEADER_SIZE 3
...@@ -61,7 +61,7 @@ static int append_string(mqtt_connection_t* connection, const char* string, int ...@@ -61,7 +61,7 @@ static int append_string(mqtt_connection_t* connection, const char* string, int
connection->buffer[connection->message.length++] = len >> 8; connection->buffer[connection->message.length++] = len >> 8;
connection->buffer[connection->message.length++] = len & 0xff; connection->buffer[connection->message.length++] = len & 0xff;
memcpy(connection->buffer + connection->message.length, string, len); c_memcpy(connection->buffer + connection->message.length, string, len);
connection->message.length += len; connection->message.length += len;
return len + 2; return len + 2;
...@@ -121,7 +121,7 @@ static mqtt_message_t* fini_message(mqtt_connection_t* connection, int type, int ...@@ -121,7 +121,7 @@ static mqtt_message_t* fini_message(mqtt_connection_t* connection, int type, int
void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length) void mqtt_msg_init(mqtt_connection_t* connection, uint8_t* buffer, uint16_t buffer_length)
{ {
memset(connection, 0, sizeof(connection)); c_memset(connection, 0, sizeof(connection));
connection->buffer = buffer; connection->buffer = buffer;
connection->buffer_length = buffer_length; connection->buffer_length = buffer_length;
} }
...@@ -294,7 +294,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf ...@@ -294,7 +294,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
variable_header->lengthMsb = 0; variable_header->lengthMsb = 0;
variable_header->lengthLsb = 4; variable_header->lengthLsb = 4;
memcpy(variable_header->magic, "MQTT", 4); c_memcpy(variable_header->magic, "MQTT", 4);
variable_header->version = 4; variable_header->version = 4;
variable_header->flags = 0; variable_header->flags = 0;
variable_header->keepaliveMsb = info->keepalive >> 8; variable_header->keepaliveMsb = info->keepalive >> 8;
...@@ -305,7 +305,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf ...@@ -305,7 +305,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
if(info->client_id != NULL && info->client_id[0] != '\0') if(info->client_id != NULL && info->client_id[0] != '\0')
{ {
if(append_string(connection, info->client_id, strlen(info->client_id)) < 0) if(append_string(connection, info->client_id, c_strlen(info->client_id)) < 0)
return fail_message(connection); return fail_message(connection);
} }
else else
...@@ -313,10 +313,10 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf ...@@ -313,10 +313,10 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
if(info->will_topic != NULL && info->will_topic[0] != '\0') if(info->will_topic != NULL && info->will_topic[0] != '\0')
{ {
if(append_string(connection, info->will_topic, strlen(info->will_topic)) < 0) if(append_string(connection, info->will_topic, c_strlen(info->will_topic)) < 0)
return fail_message(connection); return fail_message(connection);
if(append_string(connection, info->will_message, strlen(info->will_message)) < 0) if(append_string(connection, info->will_message, c_strlen(info->will_message)) < 0)
return fail_message(connection); return fail_message(connection);
variable_header->flags |= MQTT_CONNECT_FLAG_WILL; variable_header->flags |= MQTT_CONNECT_FLAG_WILL;
...@@ -327,7 +327,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf ...@@ -327,7 +327,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
if(info->username != NULL && info->username[0] != '\0') if(info->username != NULL && info->username[0] != '\0')
{ {
if(append_string(connection, info->username, strlen(info->username)) < 0) if(append_string(connection, info->username, c_strlen(info->username)) < 0)
return fail_message(connection); return fail_message(connection);
variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME; variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME;
...@@ -335,7 +335,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf ...@@ -335,7 +335,7 @@ mqtt_message_t* mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_inf
if(info->password != NULL && info->password[0] != '\0') if(info->password != NULL && info->password[0] != '\0')
{ {
if(append_string(connection, info->password, strlen(info->password)) < 0) if(append_string(connection, info->password, c_strlen(info->password)) < 0)
return fail_message(connection); return fail_message(connection);
variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD; variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD;
...@@ -351,7 +351,7 @@ mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topi ...@@ -351,7 +351,7 @@ mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topi
if(topic == NULL || topic[0] == '\0') if(topic == NULL || topic[0] == '\0')
return fail_message(connection); return fail_message(connection);
if(append_string(connection, topic, strlen(topic)) < 0) if(append_string(connection, topic, c_strlen(topic)) < 0)
return fail_message(connection); return fail_message(connection);
if(qos > 0) if(qos > 0)
...@@ -364,7 +364,7 @@ mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topi ...@@ -364,7 +364,7 @@ mqtt_message_t* mqtt_msg_publish(mqtt_connection_t* connection, const char* topi
if(connection->message.length + data_length > connection->buffer_length) if(connection->message.length + data_length > connection->buffer_length)
return fail_message(connection); return fail_message(connection);
memcpy(connection->buffer + connection->message.length, data, data_length); c_memcpy(connection->buffer + connection->message.length, data, data_length);
connection->message.length += data_length; connection->message.length += data_length;
return fini_message(connection, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain); return fini_message(connection, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain);
...@@ -412,7 +412,7 @@ mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* to ...@@ -412,7 +412,7 @@ mqtt_message_t* mqtt_msg_subscribe(mqtt_connection_t* connection, const char* to
if((*message_id = append_message_id(connection, 0)) == 0) if((*message_id = append_message_id(connection, 0)) == 0)
return fail_message(connection); return fail_message(connection);
if(append_string(connection, topic, strlen(topic)) < 0) if(append_string(connection, topic, c_strlen(topic)) < 0)
return fail_message(connection); return fail_message(connection);
if(connection->message.length + 1 > connection->buffer_length) if(connection->message.length + 1 > connection->buffer_length)
...@@ -432,7 +432,7 @@ mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char* ...@@ -432,7 +432,7 @@ mqtt_message_t* mqtt_msg_unsubscribe(mqtt_connection_t* connection, const char*
if((*message_id = append_message_id(connection, 0)) == 0) if((*message_id = append_message_id(connection, 0)) == 0)
return fail_message(connection); return fail_message(connection);
if(append_string(connection, topic, strlen(topic)) < 0) if(append_string(connection, topic, c_strlen(topic)) < 0)
return fail_message(connection); return fail_message(connection);
return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0); return fini_message(connection, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0);
......
#include "c_string.h"
#include "c_stdlib.h"
#include "c_stdio.h"
#include "msg_queue.h"
msg_queue_t *msg_enqueue(msg_queue_t **head, mqtt_message_t *msg, uint16_t msg_id, int msg_type, int publish_qos){
if(!head){
return NULL;
}
if (!msg || !msg->data || msg->length == 0){
NODE_DBG("empty message\n");
return NULL;
}
msg_queue_t *node = (msg_queue_t *)c_zalloc(sizeof(msg_queue_t));
if(!node){
NODE_DBG("not enough memory\n");
return NULL;
}
node->msg.data = (uint8_t *)c_zalloc(msg->length);
if(!node->msg.data){
NODE_DBG("not enough memory\n");
c_free(node);
return NULL;
}
c_memcpy(node->msg.data, msg->data, msg->length);
node->msg.length = msg->length;
node->next = NULL;
node->msg_id = msg_id;
node->msg_type = msg_type;
node->publish_qos = publish_qos;
msg_queue_t *tail = *head;
if(tail){
while(tail->next!=NULL) tail = tail->next;
tail->next = node;
} else {
*head = node;
}
return node;
}
void msg_destroy(msg_queue_t *node){
if(!node) return;
if(node->msg.data){
c_free(node->msg.data);
node->msg.data = NULL;
}
c_free(node);
}
msg_queue_t * msg_dequeue(msg_queue_t **head){
if(!head || !*head){
return NULL;
}
msg_queue_t *node = *head; // fetch head.
*head = node->next; // update head.
node->next = NULL;
return node;
}
#ifndef _MSG_QUEUE_H
#define _MSG_QUEUE_H 1
#include "mqtt_msg.h"
#ifdef __cplusplus
extern "C" {
#endif
struct msg_queue_t;
typedef struct msg_queue_t {
struct msg_queue_t *next;
mqtt_message_t msg;
uint16_t msg_id;
int msg_type;
int publish_qos;
} msg_queue_t;
msg_queue_t * msg_enqueue(msg_queue_t **head, mqtt_message_t *msg, uint16_t msg_id, int msg_type, int publish_qos);
void msg_destroy(msg_queue_t *node);
msg_queue_t * msg_dequeue(msg_queue_t **head);
#ifdef __cplusplus
}
#endif
#endif
...@@ -396,3 +396,33 @@ string.gsub("abc%0Ddef", "%%(%x%x)", ex) print("hello") ...@@ -396,3 +396,33 @@ string.gsub("abc%0Ddef", "%%(%x%x)", ex) print("hello")
v="abc%0D%0Adef" v="abc%0D%0Adef"
pcall(function() print(string.gsub(v, "%%(%x%x)", function(x) return string.char(tonumber(x, 16)) end)) end) pcall(function() print(string.gsub(v, "%%(%x%x)", function(x) return string.char(tonumber(x, 16)) end)) end)
m=mqtt.Client()
m:connect("192.168.18.88",1883)
topic={}
topic["/topic1"]=0
topic["/topic2"]=0
m:subscribe(topic,function(m) print("sub done") end)
m:on("message",function(m,t,pl) print(t..":") if pl~=nil then print(pl) end end )
m:publish("/topic1","hello",0,0)
m:publish("/topic3","hello",0,0) m:publish("/topic4","hello",0,0)
m=mqtt.Client()
m:connect("192.168.18.88",1883)
m:subscribe("/topic1",0,function(m) print("sub done") end)
m:subscribe("/topic2",0,function(m) print("sub done") end)
m:on("message",function(m,t,pl) print(t..":") if pl~=nil then print(pl) end end )
m:publish("/topic1","hello",0,0)
m:publish("/topic3","hello",0,0) m:publish("/topic4","hello",0,0)
m:publish("/topic1","hello1",0,0) m:publish("/topic2","hello2",0,0)
m:publish("/topic1","hello",1,0)
m:subscribe("/topic3",2,function(m) print("sub done") end)
m:publish("/topic3","hello3",2,0)
m=mqtt.Client()
m:connect("192.168.18.88",1883, function(con) print("connected hello") end)
m=mqtt.Client()
m:on("connect",function(m) print("connection") end )
m:connect("192.168.18.88",1883)
m:on("offline",function(m) print("disconnection") end )
...@@ -5,7 +5,7 @@ MEMORY ...@@ -5,7 +5,7 @@ MEMORY
dport0_0_seg : org = 0x3FF00000, len = 0x10 dport0_0_seg : org = 0x3FF00000, len = 0x10
dram0_0_seg : org = 0x3FFE8000, len = 0x14000 dram0_0_seg : org = 0x3FFE8000, len = 0x14000
iram1_0_seg : org = 0x40100000, len = 0x8000 iram1_0_seg : org = 0x40100000, len = 0x8000
irom0_0_seg : org = 0x40210000, len = 0x5A000 irom0_0_seg : org = 0x40210000, len = 0x60000
} }
PHDRS PHDRS
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment