version 1.1.2.2, 2011/11/21 14:40:56
|
version 1.1.2.8, 2011/12/06 09:04:55
|
Line 12
|
Line 12
|
* @Dup = Duplicate message |
* @Dup = Duplicate message |
* @QOS = QoS |
* @QOS = QoS |
* @Retain = Retain message |
* @Retain = Retain message |
|
* @pData = Publish data into topic |
|
* @datlen = Publish data length |
* return: -1 error or >-1 message size for send |
* return: -1 error or >-1 message size for send |
*/ |
*/ |
int |
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) | u_char Dup, u_char QOS, u_char Retain, const void *pData, int datlen) |
{ |
{ |
int siz = 0; |
int siz = 0; |
struct mqtthdr *hdr; |
struct mqtthdr *hdr; |
mqtthdr_var_t *topic; |
mqtthdr_var_t *topic; |
mqtt_v_t *mid; |
mqtt_v_t *mid; |
|
void *data; |
|
|
if (!buf || !csTopic) |
if (!buf || !csTopic) |
return -1; |
return -1; |
Line 42 mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha
|
Line 45 mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha
|
} |
} |
|
|
/* variable header */ |
/* variable header */ |
|
|
topic = (mqtthdr_var_t*) (buf->msg_base + siz); |
topic = (mqtthdr_var_t*) (buf->msg_base + siz); |
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)); |
Line 52 mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha
|
Line 54 mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha
|
mid->val = htons(msgID); |
mid->val = htons(msgID); |
siz += sizeof(mqtt_v_t); |
siz += sizeof(mqtt_v_t); |
|
|
/* fixed header */ | /* load with data */ |
| if (pData && datlen) { |
| data = buf->msg_base + siz; |
| memcpy(data, pData, datlen); |
| siz += datlen; |
| } |
|
|
|
/* fixed header */ |
|
MQTTHDR_MSGINIT(hdr); |
hdr->mqtt_msg.type = MQTT_TYPE_PUBLISH; |
hdr->mqtt_msg.type = MQTT_TYPE_PUBLISH; |
hdr->mqtt_msg.qos = QOS; |
hdr->mqtt_msg.qos = QOS; |
hdr->mqtt_msg.dup = Dup ? 1 : 0; |
hdr->mqtt_msg.dup = Dup ? 1 : 0; |
Line 84 _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd,
|
Line 93 _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd,
|
} |
} |
|
|
/* fixed header */ |
/* fixed header */ |
|
MQTTHDR_MSGINIT(hdr); |
hdr->mqtt_msg.type = cmd; |
hdr->mqtt_msg.type = cmd; |
*hdr->mqtt_len = sizeof(mqtt_v_t); |
*hdr->mqtt_len = sizeof(mqtt_v_t); |
|
|
Line 143 inline int
|
Line 153 inline int
|
mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short msgID) |
mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short msgID) |
{ |
{ |
return _mqtt_msgPUB_(buf, MQTT_TYPE_PUBCOMP, msgID); |
return _mqtt_msgPUB_(buf, MQTT_TYPE_PUBCOMP, msgID); |
|
} |
|
|
|
|
|
/* ============= decode ============ */ |
|
|
|
/* |
|
* mqtt_readPUBLISH() Read PUBLISH message |
|
* |
|
* @buf = Message buffer |
|
* @psTopic = Topic |
|
* @topicLen = Topic length |
|
* @msgID = MessageID |
|
* @pData = Data buffer |
|
* @datLen = Data buffer length, if *datLen == 0 allocate memory for pData |
|
* return: NULL error or !=NULL MQTT fixed header |
|
*/ |
|
struct mqtthdr * |
|
mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * __restrict psTopic, int topicLen, |
|
u_short *msgID, void * __restrict pData, int *datLen) |
|
{ |
|
int len, ret; |
|
struct mqtthdr *hdr; |
|
mqtthdr_var_t *var; |
|
mqtt_v_t *v; |
|
caddr_t pos; |
|
|
|
if (!buf || !psTopic || !msgID || !pData) |
|
return NULL; |
|
|
|
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBLISH, &ret, &len); |
|
if (!hdr) |
|
return NULL; |
|
if (len < sizeof(mqtt_v_t)) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return NULL; |
|
} else { |
|
pos = buf->msg_base + ret + 1; |
|
var = (mqtthdr_var_t*) pos; |
|
} |
|
|
|
/* topic */ |
|
len -= MQTTHDR_VAR_SIZEOF(var); |
|
if (len < 0) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return NULL; |
|
} else { |
|
memset(psTopic, 0, topicLen--); |
|
memcpy(psTopic, var->var_data, ntohs(var->var_sb.val) > topicLen ? |
|
topicLen : ntohs(var->var_sb.val)); |
|
pos += MQTTHDR_VAR_SIZEOF(var); |
|
v = (mqtt_v_t*) pos; |
|
} |
|
|
|
len -= sizeof(mqtt_v_t); |
|
if (len < 0) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return NULL; |
|
} else { |
|
*msgID = ntohs(v->val); |
|
pos += sizeof(mqtt_v_t); |
|
} |
|
|
|
/* data */ |
|
if (len < 0) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return NULL; |
|
} else { |
|
if (!*datLen) { |
|
if (!(pData = malloc(len))) { |
|
LOGERR; |
|
return NULL; |
|
} else |
|
*datLen = len; |
|
} |
|
|
|
memset(pData, 0, *datLen); |
|
if (len < *datLen) |
|
*datLen = len; |
|
memcpy(pData, pos, *datLen); |
|
} |
|
|
|
return hdr; |
|
} |
|
|
|
/* |
|
* mqtt_readPUBACK() Read PUBACK message |
|
* |
|
* @buf = Message buffer |
|
* return: -1 error or MessageID |
|
*/ |
|
u_short |
|
mqtt_readPUBACK(mqtt_msg_t * __restrict buf) |
|
{ |
|
int len, ret; |
|
struct mqtthdr *hdr; |
|
mqtt_v_t *v; |
|
caddr_t pos; |
|
|
|
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBACK, &ret, &len); |
|
if (!hdr) |
|
return (u_short) -1; |
|
if (len < sizeof(mqtt_v_t)) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return (u_short) -1; |
|
} else { |
|
pos = buf->msg_base + ret + 1; |
|
v = (mqtt_v_t*) pos; |
|
} |
|
|
|
return ntohs(v->val); |
|
} |
|
|
|
/* |
|
* mqtt_readPUBREC() Read PUBREC message |
|
* |
|
* @buf = Message buffer |
|
* return: -1 error or MessageID |
|
*/ |
|
u_short |
|
mqtt_readPUBREC(mqtt_msg_t * __restrict buf) |
|
{ |
|
int len, ret; |
|
struct mqtthdr *hdr; |
|
mqtt_v_t *v; |
|
caddr_t pos; |
|
|
|
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBREC, &ret, &len); |
|
if (!hdr) |
|
return (u_short) -1; |
|
if (len < sizeof(mqtt_v_t)) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return (u_short) -1; |
|
} else { |
|
pos = buf->msg_base + ret + 1; |
|
v = (mqtt_v_t*) pos; |
|
} |
|
|
|
return ntohs(v->val); |
|
} |
|
|
|
/* |
|
* mqtt_readPUBREL() Read PUBREL message |
|
* |
|
* @buf = Message buffer |
|
* return: -1 error or MessageID |
|
*/ |
|
u_short |
|
mqtt_readPUBREL(mqtt_msg_t * __restrict buf) |
|
{ |
|
int len, ret; |
|
struct mqtthdr *hdr; |
|
mqtt_v_t *v; |
|
caddr_t pos; |
|
|
|
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBREL, &ret, &len); |
|
if (!hdr) |
|
return (u_short) -1; |
|
if (len < sizeof(mqtt_v_t)) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return (u_short) -1; |
|
} else { |
|
pos = buf->msg_base + ret + 1; |
|
v = (mqtt_v_t*) pos; |
|
} |
|
|
|
return ntohs(v->val); |
|
} |
|
|
|
/* |
|
* mqtt_readPUBCOMP() Read PUBCOMP message |
|
* |
|
* @buf = Message buffer |
|
* return: -1 error or MessageID |
|
*/ |
|
u_short |
|
mqtt_readPUBCOMP(mqtt_msg_t * __restrict buf) |
|
{ |
|
int len, ret; |
|
struct mqtthdr *hdr; |
|
mqtt_v_t *v; |
|
caddr_t pos; |
|
|
|
hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBCOMP, &ret, &len); |
|
if (!hdr) |
|
return (u_short) -1; |
|
if (len < sizeof(mqtt_v_t)) { |
|
mqtt_SetErr(EINVAL, "Error:: short message length %d", len); |
|
return (u_short) -1; |
|
} else { |
|
pos = buf->msg_base + ret + 1; |
|
v = (mqtt_v_t*) pos; |
|
} |
|
|
|
return ntohs(v->val); |
} |
} |