version 1.1.2.5, 2011/11/22 10:33:39
|
version 1.1.2.8, 2011/11/28 10:17:12
|
Line 43 mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_su
|
Line 43 mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_su
|
} |
} |
|
|
/* variable header */ |
/* variable header */ |
|
|
mid = (mqtt_v_t*) (buf->msg_base + siz); |
mid = (mqtt_v_t*) (buf->msg_base + siz); |
mid->val = htons(msgID); |
mid->val = htons(msgID); |
siz += sizeof(mqtt_v_t); |
siz += sizeof(mqtt_v_t); |
|
|
/* payload with subscriptions */ |
/* payload with subscriptions */ |
| for (t = Topics; t && t->sub_topic._base; t++) { |
for (t = Topics; t && t->sub_data; t++) { | |
topic = (mqtthdr_var_t*) (buf->msg_base + siz); |
topic = (mqtthdr_var_t*) (buf->msg_base + siz); |
topic->var_sb.val = t->sub_sb.val; | topic->var_sb.val = htons(t->sub_topic._size); |
memcpy(topic->var_data, t->sub_data, ntohs(topic->var_sb.val)); | memcpy(topic->var_data, t->sub_topic._base, ntohs(topic->var_sb.val)); |
siz += MQTTHDR_VAR_SIZEOF(topic); |
siz += MQTTHDR_VAR_SIZEOF(topic); |
qos = (buf->msg_base + siz); |
qos = (buf->msg_base + siz); |
*qos = t->sub_qos; | *qos = t->sub_ret; |
siz++; |
siz++; |
} |
} |
|
|
/* fixed header */ |
/* fixed header */ |
| MQTTHDR_MSGINIT(hdr); |
hdr->mqtt_msg.type = MQTT_TYPE_SUBSCRIBE; |
hdr->mqtt_msg.type = MQTT_TYPE_SUBSCRIBE; |
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 69 mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_su
|
Line 67 mqtt_msgSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_su
|
*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); |
*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); |
|
|
mqtt_msgRealloc(buf, siz); |
mqtt_msgRealloc(buf, siz); |
|
return siz; |
|
} |
|
|
|
/* |
|
* mqtt_msgSUBACK() Create SUBACK message |
|
* |
|
* @buf = Message buffer |
|
* @Topics = MQTT subscription topics |
|
* @msgID = MessageID |
|
* return: -1 error or >-1 message size for send |
|
*/ |
|
int |
|
mqtt_msgSUBACK(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, u_short msgID) |
|
{ |
|
int siz = 0; |
|
struct mqtthdr *hdr; |
|
mqtt_v_t *v; |
|
mqtt_subscr_t *t; |
|
u_char *qos; |
|
|
|
if (!buf || !Topics) |
|
return -1; |
|
|
|
if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) |
|
return -1; |
|
else { |
|
hdr = (struct mqtthdr *) (buf->msg_base + siz); |
|
siz += sizeof(struct mqtthdr); |
|
v = (mqtt_v_t*) (buf->msg_base + siz); |
|
siz += sizeof(mqtt_v_t); |
|
} |
|
|
|
/* MessageID */ |
|
v->val = htons(msgID); |
|
|
|
/* QoS payload from subscriptions */ |
|
for (t = Topics; t && t->sub_topic._base; t++) { |
|
qos = (buf->msg_base + siz); |
|
*qos = t->sub_ret; |
|
siz++; |
|
} |
|
|
|
/* fixed header */ |
|
MQTTHDR_MSGINIT(hdr); |
|
hdr->mqtt_msg.type = MQTT_TYPE_SUBACK; |
|
*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); |
|
|
|
mqtt_msgRealloc(buf, siz); |
|
return siz; |
|
} |
|
|
|
/* |
|
* mqtt_msgUNSUBSCRIBE() Create UNSUBSCRIBE message |
|
* |
|
* @buf = Message buffer |
|
* @Topics = MQTT subscription topics |
|
* @msgID = MessageID |
|
* @Dup = Duplicate message |
|
* @QOS = QoS |
|
* return: -1 error or >-1 message size for send |
|
*/ |
|
int |
|
mqtt_msgUNSUBSCRIBE(mqtt_msg_t * __restrict buf, mqtt_subscr_t * __restrict Topics, |
|
u_short msgID, u_char Dup, u_char QOS) |
|
{ |
|
int siz = 0; |
|
struct mqtthdr *hdr; |
|
mqtthdr_var_t *topic; |
|
mqtt_v_t *mid; |
|
mqtt_subscr_t *t; |
|
|
|
if (!buf || !Topics) |
|
return -1; |
|
if (QOS > MQTT_QOS_EXACTLY) { |
|
mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter"); |
|
return -1; |
|
} |
|
if (!msgID && QOS != MQTT_QOS_ONCE) { |
|
mqtt_SetErr(EINVAL, "Error:: invalid MessageID parameter must be >0"); |
|
return -1; |
|
} |
|
|
|
if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) |
|
return -1; |
|
else { |
|
hdr = (struct mqtthdr *) (buf->msg_base + siz); |
|
siz += sizeof(struct mqtthdr); |
|
} |
|
|
|
/* variable header */ |
|
mid = (mqtt_v_t*) (buf->msg_base + siz); |
|
mid->val = htons(msgID); |
|
siz += sizeof(mqtt_v_t); |
|
|
|
/* payload with subscriptions */ |
|
for (t = Topics; t && t->sub_topic._base; t++) { |
|
topic = (mqtthdr_var_t*) (buf->msg_base + siz); |
|
topic->var_sb.val = htons(t->sub_topic._size); |
|
memcpy(topic->var_data, t->sub_topic._base, ntohs(topic->var_sb.val)); |
|
siz += MQTTHDR_VAR_SIZEOF(topic); |
|
} |
|
|
|
/* fixed header */ |
|
MQTTHDR_MSGINIT(hdr); |
|
hdr->mqtt_msg.type = MQTT_TYPE_UNSUBSCRIBE; |
|
hdr->mqtt_msg.qos = QOS; |
|
hdr->mqtt_msg.dup = Dup ? 1 : 0; |
|
hdr->mqtt_msg.retain = 0; |
|
*hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); |
|
|
|
mqtt_msgRealloc(buf, siz); |
|
return siz; |
|
} |
|
|
|
/* |
|
* mqtt_msgUNSUBACK() Create UNSUBACK message |
|
* |
|
* @buf = Message buffer |
|
* @msgID = MessageID |
|
* return: -1 error or >-1 message size for send |
|
*/ |
|
int |
|
mqtt_msgUNSUBACK(mqtt_msg_t * __restrict buf, u_short msgID) |
|
{ |
|
int siz = 0; |
|
struct mqtthdr *hdr; |
|
mqtt_v_t *v; |
|
|
|
if (!buf) |
|
return -1; |
|
|
|
if (mqtt_msgRealloc(buf, sizeof(struct mqtthdr) + sizeof(mqtt_v_t)) == -1) |
|
return -1; |
|
else { |
|
hdr = (struct mqtthdr *) (buf->msg_base + siz); |
|
siz += sizeof(struct mqtthdr); |
|
v = (mqtt_v_t*) (buf->msg_base + siz); |
|
siz += sizeof(mqtt_v_t); |
|
} |
|
|
|
/* fixed header */ |
|
MQTTHDR_MSGINIT(hdr); |
|
hdr->mqtt_msg.type = MQTT_TYPE_UNSUBACK; |
|
*hdr->mqtt_len = sizeof(mqtt_v_t); |
|
|
|
/* MessageID */ |
|
v->val = htons(msgID); |
|
|
return siz; |
return siz; |
} |
} |