--- libaitmqtt/src/pub.c 2012/01/26 13:07:33 1.1.1.1 +++ libaitmqtt/src/pub.c 2012/04/27 15:49:07 1.1.1.1.2.4 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: pub.c,v 1.1.1.1 2012/01/26 13:07:33 misho Exp $ +* $Id: pub.c,v 1.1.1.1.2.4 2012/04/27 15:49:07 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 +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -46,8 +46,6 @@ SUCH DAMAGE. #include "global.h" -/* ------------------------------------------------------------------- */ - /* * mqtt_msgPUBLISH() Create PUBLISH message * @@ -74,11 +72,11 @@ mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha if (!buf || !csTopic) return -1; if (QOS > MQTT_QOS_EXACTLY) { - mqtt_SetErr(EINVAL, "Error:: invalid QoS parameter"); + mqtt_SetErr(EINVAL, "Invalid QoS parameter"); return -1; } 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; } @@ -114,7 +112,6 @@ mqtt_msgPUBLISH(mqtt_msg_t * __restrict buf, const cha hdr->mqtt_msg.retain = Retain ? 1 : 0; *hdr->mqtt_len = mqtt_encodeLen(siz - sizeof(struct mqtthdr)); - mqtt_msgRealloc(buf, siz); return siz; } @@ -211,12 +208,11 @@ mqtt_msgPUBCOMP(mqtt_msg_t * __restrict buf, u_short m * @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 + * return: -1 error or !=-1 allocated data buffer length */ -struct mqtthdr * +int 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; struct mqtthdr *hdr; @@ -225,19 +221,19 @@ mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _ caddr_t pos; if (!buf || !psTopic || !msgID || !pData) - return NULL; + return -1; hdr = _mqtt_readHEADER(buf, MQTT_TYPE_PUBLISH, &ret, &len); if (!hdr) - return NULL; + 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, "Error:: short message length %d", len); - return NULL; + 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 ? @@ -248,8 +244,8 @@ mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _ len -= sizeof(mqtt_len_t); if (len < 0) { - mqtt_SetErr(EINVAL, "Error:: short message length %d", len); - return NULL; + mqtt_SetErr(EINVAL, "Short message length %d", len); + return -1; } else { *msgID = ntohs(v->val); pos += sizeof(mqtt_len_t); @@ -257,24 +253,19 @@ mqtt_readPUBLISH(mqtt_msg_t * __restrict buf, char * _ /* data */ if (len < 0) { - mqtt_SetErr(EINVAL, "Error:: short message length %d", len); - return NULL; + mqtt_SetErr(EINVAL, "Short message length %d", len); + return -1; } else { - if (!*datLen) { - if (!(pData = malloc(len))) { - LOGERR; - return NULL; - } else - *datLen = len; - } + if (!(*pData = malloc(len + 1))) { + LOGERR; + return -1; + } else + ((char*) (*pData))[len] = 0; - memset(pData, 0, *datLen); - if (len < *datLen) - *datLen = len; - memcpy(pData, pos, *datLen); + memcpy(*pData, pos, len); } - return hdr; + return len; } /* @@ -295,7 +286,7 @@ mqtt_readPUBACK(mqtt_msg_t * __restrict buf) if (!hdr) return (u_short) -1; 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; } else { pos = buf->msg_base + ret + 1; @@ -323,7 +314,7 @@ mqtt_readPUBREC(mqtt_msg_t * __restrict buf) if (!hdr) return (u_short) -1; 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; } else { pos = buf->msg_base + ret + 1; @@ -351,7 +342,7 @@ mqtt_readPUBREL(mqtt_msg_t * __restrict buf) if (!hdr) return (u_short) -1; 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; } else { pos = buf->msg_base + ret + 1; @@ -379,7 +370,7 @@ mqtt_readPUBCOMP(mqtt_msg_t * __restrict buf) if (!hdr) return (u_short) -1; 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; } else { pos = buf->msg_base + ret + 1;