--- libaitmqtt/src/pub.c 2012/06/11 08:37:41 1.1.1.1.2.5 +++ libaitmqtt/src/pub.c 2022/09/14 18:36:23 1.4.4.2 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: pub.c,v 1.1.1.1.2.5 2012/06/11 08:37:41 misho Exp $ +* $Id: pub.c,v 1.4.4.2 2022/09/14 18:36:23 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -12,7 +12,7 @@ terms: All of the documentation and software included in the ELWIX and AITNET Releases is copyrighted by ELWIX - Sofia/Bulgaria -Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 +Copyright 2004 - 2022 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -49,7 +49,6 @@ SUCH DAMAGE. /* * mqtt_msgPUBLISH() Create PUBLISH message * - * @buf = Message buffer * @csTopic = Publish topic * @msgID = MessageID >0, if QOS != MQTT_QOS_ONCE * @Dup = Duplicate message @@ -57,325 +56,142 @@ SUCH DAMAGE. * @Retain = Retain message * @pData = Publish data into topic * @datlen = Publish data length - * return: -1 error or >-1 message size for send + * return: NULL error or allocated PUBLISH message */ -int -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) +mqtt_msg_t * +mqtt_msgPUBLISH(const char *csTopic, u_short msgID, 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; mqtthdr_var_t *topic; mqtt_len_t *mid; void *data; + mqtt_msg_t *msg = NULL; - if (!buf || !csTopic) - return -1; + if (!csTopic) + return NULL; if (QOS > MQTT_QOS_EXACTLY) { mqtt_SetErr(EINVAL, "Invalid QoS parameter"); - return -1; + return NULL; } if (!msgID && QOS != MQTT_QOS_ONCE) { mqtt_SetErr(EINVAL, "Invalid MessageID parameter must be >0"); - return -1; + return NULL; } - if (mqtt_msgRealloc(buf, MQTTMSG_MAX) == -1) - return -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 (!(msg = mqtt_msgAlloc(siz + len))) + return NULL; else { - hdr = (struct mqtthdr *) (buf->msg_base + siz); - siz += sizeof(struct mqtthdr); + data = msg->msg_base; + hdr = (struct mqtthdr *) data; } + /* fixed header */ + 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 */ - topic = (mqtthdr_var_t*) (buf->msg_base + siz); + topic = (mqtthdr_var_t*) data; topic->var_sb.val = htons(strlen(csTopic)); 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); - siz += sizeof(mqtt_len_t); + data += sizeof(mqtt_len_t); /* load with data */ - if (pData && datlen) { - data = buf->msg_base + siz; + if (pData && datlen) 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; - hdr->mqtt_msg.retain = Retain ? 1 : 0; - *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); - - return siz; + return msg; } -static int -_mqtt_msgPUB_(mqtt_msg_t * __restrict buf, u_char cmd, u_short msgID) +static mqtt_msg_t * +_mqtt_msgPUB_(u_char cmd, u_short msgID) { - int siz = 0; struct mqtthdr *hdr; mqtt_len_t *v; + mqtt_msg_t *msg = NULL; - if (!buf) - return -1; - - if (mqtt_msgRealloc(buf, sizeof(struct mqtthdr) + sizeof(mqtt_len_t)) == -1) - return -1; + if (!(msg = mqtt_msgAlloc(sizeof(struct mqtthdr) + sizeof(mqtt_len_t)))) + return NULL; else { - hdr = (struct mqtthdr *) (buf->msg_base + siz); - siz += sizeof(struct mqtthdr); - v = (mqtt_len_t*) (buf->msg_base + siz); - siz += sizeof(mqtt_len_t); + hdr = (struct mqtthdr *) msg->msg_base; + v = (mqtt_len_t*) (msg->msg_base + sizeof(struct mqtthdr)); } /* fixed header */ - MQTTHDR_MSGINIT(hdr); hdr->mqtt_msg.type = cmd; *hdr->mqtt_len = sizeof(mqtt_len_t); /* MessageID */ v->val = htons(msgID); - return siz; + return msg; } /* * mqtt_msgPUBACK() Create PUBACK message * - * @buf = Message buffer * @msgID = MessageID - * return: -1 error or >-1 message size for send + * return: NULL error or allocated PUBACK message */ -inline int -mqtt_msgPUBACK(mqtt_msg_t * __restrict buf, u_short msgID) +mqtt_msg_t * +mqtt_msgPUBACK(u_short msgID) { - return _mqtt_msgPUB_(buf, MQTT_TYPE_PUBACK, msgID); + return _mqtt_msgPUB_(MQTT_TYPE_PUBACK, msgID); } /* * mqtt_msgPUBREC() Create PUBREC message * - * @buf = Message buffer * @msgID = MessageID - * return: -1 error or >-1 message size for send + * return: NULL error or allocated PUBREC message */ -inline int -mqtt_msgPUBREC(mqtt_msg_t * __restrict buf, u_short msgID) +mqtt_msg_t * +mqtt_msgPUBREC(u_short msgID) { - return _mqtt_msgPUB_(buf, MQTT_TYPE_PUBREC, msgID); + return _mqtt_msgPUB_(MQTT_TYPE_PUBREC, msgID); } /* * mqtt_msgPUBREL() Create PUBREL message * - * @buf = Message buffer * @msgID = MessageID - * return: -1 error or >-1 message size for send + * return: NULL error or allocated PUBREL message */ -inline int -mqtt_msgPUBREL(mqtt_msg_t * __restrict buf, u_short msgID) +mqtt_msg_t * +mqtt_msgPUBREL(u_short msgID) { - return _mqtt_msgPUB_(buf, MQTT_TYPE_PUBREL, msgID); + return _mqtt_msgPUB_(MQTT_TYPE_PUBREL, msgID); } /* * mqtt_msgPUBCOMP() Create PUBCOMP message * - * @buf = Message buffer * @msgID = MessageID - * return: -1 error or >-1 message size for send + * return: NULL error or allocated PUBCOMP message */ -inline int -mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short msgID) +mqtt_msg_t * +mqtt_msgPUBCOMP(u_short 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, may be NULL - * return: -1 error or !=-1 allocated data buffer length - */ -int -mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * __restrict psTopic, int topicLen, - u_short *msgID, void ** __restrict pData) -{ - int len, ret; - struct mqtthdr *hdr; - mqtthdr_var_t *var; - mqtt_len_t *v; - caddr_t pos; - - if (!buf || !psTopic || !msgID) - return -1; - - hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBLISH, &ret, &len); - if (!hdr) - return -1; - pos = buf->msg_base + ret + 1; - var = (mqtthdr_var_t*) pos; - - /* topic */ - len -= MQTTHDR_VAR_SIZEOF(var); - if (len < 0) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return -1; - } 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_len_t*) pos; - } - - len -= sizeof(mqtt_len_t); - if (len < 0) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return -1; - } else { - *msgID = ntohs(v->val); - pos += sizeof(mqtt_len_t); - } - - /* data */ - if (len < 0) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return -1; - } else if (pData) { - if (!(*pData = malloc(len + 1))) { - LOGERR; - return -1; - } else - ((char*) (*pData))[len] = 0; - - memcpy(*pData, pos, len); - } - - return len; -} - -/* - * 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_len_t *v; - caddr_t pos; - - hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBACK, &ret, &len); - if (!hdr) - return (u_short) -1; - if (len < sizeof(mqtt_len_t)) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return (u_short) -1; - } else { - pos = buf->msg_base + ret + 1; - v = (mqtt_len_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_len_t *v; - caddr_t pos; - - hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBREC, &ret, &len); - if (!hdr) - return (u_short) -1; - if (len < sizeof(mqtt_len_t)) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return (u_short) -1; - } else { - pos = buf->msg_base + ret + 1; - v = (mqtt_len_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_len_t *v; - caddr_t pos; - - hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBREL, &ret, &len); - if (!hdr) - return (u_short) -1; - if (len < sizeof(mqtt_len_t)) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return (u_short) -1; - } else { - pos = buf->msg_base + ret + 1; - v = (mqtt_len_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_len_t *v; - caddr_t pos; - - hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBCOMP, &ret, &len); - if (!hdr) - return (u_short) -1; - if (len < sizeof(mqtt_len_t)) { - mqtt_SetErr(EINVAL, "Short message length %d", len); - return (u_short) -1; - } else { - pos = buf->msg_base + ret + 1; - v = (mqtt_len_t*) pos; - } - - return ntohs(v->val); + return _mqtt_msgPUB_(MQTT_TYPE_PUBCOMP, msgID); }