--- mqtt/src/Attic/pub.c 2011/11/21 14:40:56 1.1.2.2 +++ mqtt/src/Attic/pub.c 2011/12/05 22:17:39 1.1.2.7 @@ -12,16 +12,19 @@ * @Dup = Duplicate message * @QOS = QoS * @Retain = Retain message + * @pData = Publish data into topic + * @datlen = Publish data length * return: -1 error or >-1 message size for send */ int 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, u_short datlen) { int siz = 0; struct mqtthdr *hdr; mqtthdr_var_t *topic; mqtt_v_t *mid; + void *data; if (!buf || !csTopic) return -1; @@ -42,7 +45,6 @@ mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha } /* variable header */ - topic = (mqtthdr_var_t*) (buf->msg_base + siz); topic->var_sb.val = htons(strlen(csTopic)); memcpy(topic->var_data, csTopic, ntohs(topic->var_sb.val)); @@ -52,8 +54,15 @@ mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha mid->val = htons(msgID); 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.qos = QOS; hdr->mqtt_msg.dup = Dup ? 1 : 0; @@ -84,6 +93,7 @@ _mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd, } /* fixed header */ + MQTTHDR_MSGINIT(hdr); hdr->mqtt_msg.type = cmd; *hdr->mqtt_len = sizeof(mqtt_v_t); @@ -143,4 +153,119 @@ inline int mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short msgID) { return _mqtt_msgPUB_(buf, MQTT_TYPE_PUBCOMP, msgID); +} + + +/* ============= decode ============ */ + +/* + * 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); }