version 1.1.1.1, 2012/01/26 13:07:33
|
version 1.2.2.2, 2012/06/28 09:01:42
|
Line 12 terms:
|
Line 12 terms:
|
All of the documentation and software included in the ELWIX and AITNET |
All of the documentation and software included in the ELWIX and AITNET |
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org> |
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org> |
|
|
Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011 | Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 |
by Michael Pounov <misho@elwix.org>. All rights reserved. |
by Michael Pounov <misho@elwix.org>. All rights reserved. |
|
|
Redistribution and use in source and binary forms, with or without |
Redistribution and use in source and binary forms, with or without |
Line 46 SUCH DAMAGE.
|
Line 46 SUCH DAMAGE.
|
#include "global.h" |
#include "global.h" |
|
|
|
|
/* ------------------------------------------------------------------- */ |
|
|
|
/* |
/* |
* mqtt_msgPUBLISH() Create PUBLISH message |
* mqtt_msgPUBLISH() Create PUBLISH message |
* |
* |
Line 65 int
|
Line 63 int
|
mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const char *csTopic, u_short msgID, |
mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const char *csTopic, u_short msgID, |
u_char Dup, u_char QOS, u_char Retain, const void *pData, int datlen) |
u_char Dup, u_char QOS, u_char Retain, const void *pData, int datlen) |
{ |
{ |
int siz = 0; | int len, siz; |
| u_int n, *l; |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
mqtthdr_var_t *topic; |
mqtthdr_var_t *topic; |
mqtt_len_t *mid; |
mqtt_len_t *mid; |
Line 74 mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha
|
Line 73 mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha
|
if (!buf || !csTopic) |
if (!buf || !csTopic) |
return -1; |
return -1; |
if (QOS > MQTT_QOS_EXACTLY) { |
if (QOS > MQTT_QOS_EXACTLY) { |
mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter"); | mqtt_SetErr(EINVAL, "Invalid QoS parameter"); |
return -1; |
return -1; |
} |
} |
if (!msgID && QOS != MQTT_QOS_ONCE) { |
if (!msgID && QOS != MQTT_QOS_ONCE) { |
mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0"); | mqtt_SetErr(EINVAL, "Invalid MessageID parameter must be >0"); |
return -1; |
return -1; |
} |
} |
|
|
if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) | /* calculate message size */ |
| len = sizeof(mqtt_len_t) + strlen(csTopic); /* topic */ |
| len += sizeof(mqtt_len_t); /* msgid */ |
| len += datlen; /* data len */ |
| |
| /* calculate header size */ |
| siz = sizeof(struct mqtthdr); /* mqtt fixed header */ |
| n = mqtt_encodeLen(len); /* message size */ |
| siz += mqtt_sizeLen(n) - 1; /* length size */ |
| |
| if (mqtt_msgRealloc(buf, siz + len) == -1) |
return -1; |
return -1; |
else { |
else { |
hdr = (struct mqtthdr *) (buf->msg_base + siz); | data = buf->msg_base; |
siz += sizeof(struct mqtthdr); | hdr = (struct mqtthdr *) data; |
} |
} |
|
|
|
/* fixed header */ |
|
MQTTHDR_MSGINIT(hdr); |
|
hdr->mqtt_msg.type = MQTT_TYPE_PUBLISH; |
|
hdr->mqtt_msg.qos = QOS; |
|
hdr->mqtt_msg.dup = Dup ? 1 : 0; |
|
hdr->mqtt_msg.retain = Retain ? 1 : 0; |
|
l = (u_int*) hdr->mqtt_len; |
|
*l = n; |
|
data += siz; |
|
|
/* variable header */ |
/* variable header */ |
topic = (mqtthdr_var_t*) (buf->msg_base + siz); | topic = (mqtthdr_var_t*) data; |
topic->var_sb.val = htons(strlen(csTopic)); |
topic->var_sb.val = htons(strlen(csTopic)); |
memcpy(topic->var_data, csTopic, ntohs(topic->var_sb.val)); |
memcpy(topic->var_data, csTopic, ntohs(topic->var_sb.val)); |
siz += MQTTHDR_VAR_SIZEOF(topic); | data += MQTTHDR_VAR_SIZEOF(topic); |
|
|
mid = (mqtt_len_t*) (buf->msg_base + siz); | mid = (mqtt_len_t*) data; |
mid->val = htons(msgID); |
mid->val = htons(msgID); |
siz += sizeof(mqtt_len_t); | data += sizeof(mqtt_len_t); |
|
|
/* load with data */ |
/* load with data */ |
if (pData && datlen) { | if (pData && datlen) |
data = buf->msg_base + siz; | |
memcpy(data, pData, datlen); |
memcpy(data, pData, datlen); |
siz += datlen; |
|
} |
|
|
|
/* fixed header */ | return siz + len; |
MQTTHDR_MSGINIT(hdr); | |
hdr->mqtt_msg.type = MQTT_TYPE_PUBLISH; | |
hdr->mqtt_msg.qos = QOS; | |
hdr->mqtt_msg.dup = Dup ? 1 : 0; | |
hdr->mqtt_msg.retain = Retain ? 1 : 0; | |
*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); | |
| |
mqtt_msgRealloc(buf, siz); | |
return siz; | |
} |
} |
|
|
static int |
static int |
_mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd, u_short msgID) |
_mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd, u_short msgID) |
{ |
{ |
int siz = 0; | int siz = sizeof(struct mqtthdr); |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
mqtt_len_t *v; |
mqtt_len_t *v; |
|
|
Line 130 _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd,
|
Line 137 _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd,
|
|
|
if (mqtt_msgRealloc(buf, sizeof(struct mqtthdr) + sizeof(mqtt_len_t)) == -1) |
if (mqtt_msgRealloc(buf, sizeof(struct mqtthdr) + sizeof(mqtt_len_t)) == -1) |
return -1; |
return -1; |
else { | else |
hdr = (struct mqtthdr *) (buf->msg_base + siz); | hdr = (struct mqtthdr *) buf->msg_base; |
siz += sizeof(struct mqtthdr); | |
v = (mqtt_len_t*) (buf->msg_base + siz); | |
siz += sizeof(mqtt_len_t); | |
} | |
|
|
/* fixed header */ |
/* fixed header */ |
MQTTHDR_MSGINIT(hdr); |
MQTTHDR_MSGINIT(hdr); |
Line 143 _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd,
|
Line 146 _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd,
|
*hdr->mqtt_len = sizeof(mqtt_len_t); |
*hdr->mqtt_len = sizeof(mqtt_len_t); |
|
|
/* MessageID */ |
/* MessageID */ |
|
v = (mqtt_len_t*) (buf->msg_base + siz); |
v->val = htons(msgID); |
v->val = htons(msgID); |
|
siz += sizeof(mqtt_len_t); |
|
|
return siz; |
return siz; |
} |
} |
Line 210 mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short m
|
Line 215 mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short m
|
* @psTopic = Topic |
* @psTopic = Topic |
* @topicLen = Topic length |
* @topicLen = Topic length |
* @msgID = MessageID |
* @msgID = MessageID |
* @pData = Data buffer | * @pData = Data buffer, may be NULL |
* @datLen = Data buffer length, if *datLen == 0 allocate memory for pData | * return: -1 error or !=-1 allocated data buffer length |
* return: NULL error or !=NULL MQTT fixed header | |
*/ |
*/ |
struct mqtthdr * | int |
mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * __restrict psTopic, int topicLen, |
mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * __restrict psTopic, int topicLen, |
u_short *msgID, void * __restrict pData, int *datLen) | u_short *msgID, void ** __restrict pData) |
{ |
{ |
int len, ret; |
int len, ret; |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
Line 224 mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _
|
Line 228 mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _
|
mqtt_len_t *v; |
mqtt_len_t *v; |
caddr_t pos; |
caddr_t pos; |
|
|
if (!buf || !psTopic || !msgID || !pData) | if (!buf || !psTopic || !msgID) |
return NULL; | return -1; |
|
|
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBLISH, &ret, &len); |
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBLISH, &ret, &len); |
if (!hdr) |
if (!hdr) |
return NULL; | return -1; |
pos = buf->msg_base + ret + 1; |
pos = buf->msg_base + ret + 1; |
var = (mqtthdr_var_t*) pos; |
var = (mqtthdr_var_t*) pos; |
|
|
/* topic */ |
/* topic */ |
len -= MQTTHDR_VAR_SIZEOF(var); |
len -= MQTTHDR_VAR_SIZEOF(var); |
if (len < 0) { |
if (len < 0) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return NULL; | return -1; |
} else { |
} else { |
memset(psTopic, 0, topicLen--); |
memset(psTopic, 0, topicLen--); |
memcpy(psTopic, var->var_data, ntohs(var->var_sb.val) > topicLen ? |
memcpy(psTopic, var->var_data, ntohs(var->var_sb.val) > topicLen ? |
Line 248 mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _
|
Line 252 mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _
|
|
|
len -= sizeof(mqtt_len_t); |
len -= sizeof(mqtt_len_t); |
if (len < 0) { |
if (len < 0) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return NULL; | return -1; |
} else { |
} else { |
*msgID = ntohs(v->val); |
*msgID = ntohs(v->val); |
pos += sizeof(mqtt_len_t); |
pos += sizeof(mqtt_len_t); |
Line 257 mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _
|
Line 261 mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _
|
|
|
/* data */ |
/* data */ |
if (len < 0) { |
if (len < 0) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return NULL; | return -1; |
} else { | } else if (pData) { |
if (!*datLen) { | if (!(*pData = malloc(len + 1))) { |
if (!(pData = malloc(len))) { | LOGERR; |
LOGERR; | return -1; |
return NULL; | } else |
} else | ((char*) (*pData))[len] = 0; |
*datLen = len; | |
} | |
|
|
memset(pData, 0, *datLen); | memcpy(*pData, pos, len); |
if (len < *datLen) | |
*datLen = len; | |
memcpy(pData, pos, *datLen); | |
} |
} |
|
|
return hdr; | return len; |
} |
} |
|
|
/* |
/* |
Line 295 mqtt_readPUBACK(mqtt_msg_t * __restrict buf)
|
Line 294 mqtt_readPUBACK(mqtt_msg_t * __restrict buf)
|
if (!hdr) |
if (!hdr) |
return (u_short) -1; |
return (u_short) -1; |
if (len < sizeof(mqtt_len_t)) { |
if (len < sizeof(mqtt_len_t)) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return (u_short) -1; |
return (u_short) -1; |
} else { |
} else { |
pos = buf->msg_base + ret + 1; |
pos = buf->msg_base + ret + 1; |
Line 323 mqtt_readPUBREC(mqtt_msg_t * __restrict buf)
|
Line 322 mqtt_readPUBREC(mqtt_msg_t * __restrict buf)
|
if (!hdr) |
if (!hdr) |
return (u_short) -1; |
return (u_short) -1; |
if (len < sizeof(mqtt_len_t)) { |
if (len < sizeof(mqtt_len_t)) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return (u_short) -1; |
return (u_short) -1; |
} else { |
} else { |
pos = buf->msg_base + ret + 1; |
pos = buf->msg_base + ret + 1; |
Line 351 mqtt_readPUBREL(mqtt_msg_t * __restrict buf)
|
Line 350 mqtt_readPUBREL(mqtt_msg_t * __restrict buf)
|
if (!hdr) |
if (!hdr) |
return (u_short) -1; |
return (u_short) -1; |
if (len < sizeof(mqtt_len_t)) { |
if (len < sizeof(mqtt_len_t)) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return (u_short) -1; |
return (u_short) -1; |
} else { |
} else { |
pos = buf->msg_base + ret + 1; |
pos = buf->msg_base + ret + 1; |
Line 379 mqtt_readPUBCOMP(mqtt_msg_t * __restrict buf)
|
Line 378 mqtt_readPUBCOMP(mqtt_msg_t * __restrict buf)
|
if (!hdr) |
if (!hdr) |
return (u_short) -1; |
return (u_short) -1; |
if (len < sizeof(mqtt_len_t)) { |
if (len < sizeof(mqtt_len_t)) { |
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); | mqtt_SetErr(EINVAL, "Short message length %d", len); |
return (u_short) -1; |
return (u_short) -1; |
} else { |
} else { |
pos = buf->msg_base + ret + 1; |
pos = buf->msg_base + ret + 1; |