--- libaitmqtt/src/cliside.c 2022/09/15 15:04:44 1.3.12.2 +++ libaitmqtt/src/cliside.c 2022/09/15 15:48:41 1.3.12.3 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: cliside.c,v 1.3.12.2 2022/09/15 15:04:44 misho Exp $ +* $Id: cliside.c,v 1.3.12.3 2022/09/15 15:48:41 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -157,24 +157,32 @@ mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_s return NULL; /* receive suback */ + cli->buf = mqtt_msgAlloc(BUFSIZ); + if (!cli->buf) + return NULL; siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); if (siz == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return NULL; } siz = mqtt_readSUBACK(cli->buf, &mid, &qoses); - if (siz == -1) + if (siz == -1) { + mqtt_msgFree(&cli->buf, 0); return NULL; + } if (msgID != mid) { - free(qoses); + if (qoses) + e_free(qoses); + mqtt_msgFree(&cli->buf, 0); mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, mid); return NULL; } + mqtt_msgFree(&cli->buf, 0); return qoses; } -#if 0 /* * mqtt_cli_Unsubscribe() - Unsubscribe from broker * @@ -186,7 +194,7 @@ mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_s * return: -1 error or 0 ok */ int -mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t * __restrict Topics, +mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t ** __restrict Topics, u_short msgID, u_char Dup, u_char QoS) { int siz = 0; @@ -195,15 +203,16 @@ mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt return -1; /* send unsubscribe */ - siz = mqtt_msgUNSUBSCRIBE(cli->buf, Topics, msgID, Dup, QoS); - if (siz == -1) + cli->buf = mqtt_msgUNSUBSCRIBE(Topics, msgID, Dup, QoS); + if (!cli->buf) return -1; - siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL); + siz = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); if (siz == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return -1; } else - memset(cli->buf->msg_base, 0, cli->buf->msg_len); + mqtt_msgFree(&cli->buf, 0); if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { return -1; @@ -211,19 +220,27 @@ mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt return -1; /* receive unsuback */ + cli->buf = mqtt_msgAlloc(BUFSIZ); + if (!cli->buf) + return -1; siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); if (siz == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return -1; } siz = mqtt_readUNSUBACK(cli->buf); - if (siz == -1) + if (siz == -1) { + mqtt_msgFree(&cli->buf, 0); return -1; + } if (msgID != siz) { + mqtt_msgFree(&cli->buf, 0); mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); return -1; } + mqtt_msgFree(&cli->buf, 0); return 0; } @@ -250,17 +267,16 @@ mqtt_cli_Publish(mqtt_cli_t * __restrict cli, u_short return -1; /* send publish */ - siz = mqtt_msgPUBLISH(cli->buf, csTopic, msgID, Dup, QoS, Retain, pData, datLen); - if (siz == -1) + cli->buf = mqtt_msgPUBLISH(csTopic, msgID, Dup, QoS, Retain, pData, datLen); + if (!cli->buf) return -1; - siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL); - if (siz == -1) { + wlen = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); + if (wlen == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return -1; - } else { - wlen = siz; - memset(cli->buf->msg_base, 0, cli->buf->msg_len); - } + } else + mqtt_msgFree(&cli->buf, 0); if (QoS == MQTT_QOS_ONCE) /* no reply */ goto end; @@ -271,42 +287,55 @@ mqtt_cli_Publish(mqtt_cli_t * __restrict cli, u_short return -1; /* receive PUBxxx */ + cli->buf = mqtt_msgAlloc(BUFSIZ); + if (!cli->buf) + return -1; siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); if (siz == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return -1; } if (QoS == MQTT_QOS_ACK) { /* reply with PUBACK */ siz = mqtt_readPUBACK(cli->buf); - if (siz == -1) + if (siz == -1) { + mqtt_msgFree(&cli->buf, 0); return -1; + } if (msgID != siz) { + mqtt_msgFree(&cli->buf, 0); mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); return -1; } + mqtt_msgFree(&cli->buf, 0); goto end; } else { /* reply with PUBREC */ siz = mqtt_readPUBREC(cli->buf); - if (siz == -1) + if (siz == -1) { + mqtt_msgFree(&cli->buf, 0); return -1; + } if (msgID != siz) { + mqtt_msgFree(&cli->buf, 0); mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); return -1; } + mqtt_msgFree(&cli->buf, 0); } do { /* send publish release QoS == 2 */ - siz = mqtt_msgPUBREL(cli->buf, msgID); - if (siz == -1) + cli->buf = mqtt_msgPUBREL(msgID); + if (!cli->buf) return -1; - siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL); + siz = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); if (siz == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return -1; } else - memset(cli->buf->msg_base, 0, cli->buf->msg_len); + mqtt_msgFree(&cli->buf, 0); if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { return -1; @@ -318,25 +347,32 @@ mqtt_cli_Publish(mqtt_cli_t * __restrict cli, u_short } /* receive PUBCOMP */ + cli->buf = mqtt_msgAlloc(BUFSIZ); + if (!cli->buf) + return -1; siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); if (siz == -1) { LOGERR; + mqtt_msgFree(&cli->buf, 0); return -1; } siz = mqtt_readPUBCOMP(cli->buf); - if (siz == -1) + if (siz == -1) { + mqtt_msgFree(&cli->buf, 0); return -1; + } if (msgID != siz) { + mqtt_msgFree(&cli->buf, 0); mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); if (Dup++ > 1) return -1; else continue; } + mqtt_msgFree(&cli->buf, 0); } while (0); end: return wlen; } -#endif