--- libaitmqtt/src/cliside.c 2012/05/08 11:29:56 1.1.2.5 +++ libaitmqtt/src/cliside.c 2022/09/15 15:04:44 1.3.12.2 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: cliside.c,v 1.1.2.5 2012/05/08 11:29:56 misho Exp $ +* $Id: cliside.c,v 1.3.12.2 2022/09/15 15:04:44 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 @@ -61,7 +61,7 @@ mqtt_cli_Open(struct sockaddr *addr, u_short timeout) if (!addr) return NULL; - cli = malloc(sizeof(mqtt_cli_t)); + cli = e_malloc(sizeof(mqtt_cli_t)); if (!cli) { LOGERR; return NULL; @@ -72,23 +72,22 @@ mqtt_cli_Open(struct sockaddr *addr, u_short timeout) cli->sock = socket(addr->sa_family, SOCK_STREAM, IPPROTO_TCP); if (cli->sock == -1) { LOGERR; - free(cli); + e_free(cli); return NULL; } +#ifndef __linux__ if (connect(cli->sock, addr, addr->sa_len) == -1) { +#else + if (connect(cli->sock, addr, addr->sa_family == AF_INET6 ? + sizeof(struct sockaddr_in6) : + sizeof(struct sockaddr_in)) == -1) { +#endif LOGERR; close(cli->sock); - free(cli); + e_free(cli); return NULL; } - cli->buf = mqtt_msgAlloc(USHRT_MAX); - if (!cli->buf) { - close(cli->sock); - free(cli); - return NULL; - } - return cli; } @@ -101,23 +100,23 @@ mqtt_cli_Open(struct sockaddr *addr, u_short timeout) int mqtt_cli_Close(mqtt_cli_t ** __restrict cli) { - int siz = 0; + int siz; if (!cli || !*cli) return -1; /* send disconnect */ - siz = mqtt_msgDISCONNECT((*cli)->buf); - if (siz > -1) { - siz = send((*cli)->sock, (*cli)->buf->msg_base, siz, MSG_NOSIGNAL); + (*cli)->buf = mqtt_msgDISCONNECT(); + if ((*cli)->buf) { + siz = send((*cli)->sock, (*cli)->buf->msg_base, (*cli)->buf->msg_len, MSG_NOSIGNAL); if (siz > -1) shutdown((*cli)->sock, SHUT_RDWR); } close((*cli)->sock); - mqtt_msgFree(&(*cli)->buf, 42); + mqtt_msgFree(&(*cli)->buf, 0); - free(*cli); + e_free(*cli); *cli = NULL; return 0; } @@ -128,14 +127,11 @@ mqtt_cli_Close(mqtt_cli_t ** __restrict cli) * @cli = connected client * @Topics = Topics for subscribes * @msgID = Message ID - * @Dup = Duplicated request - * @QoS = Message QoS * return: NULL error or !=NULL allocated array with subscribed QoS responses, - * must be free() result! + * must be e_free() result! */ u_char * -mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t * __restrict Topics, - u_short msgID, u_char Dup, u_char QoS) +mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t ** __restrict Topics, u_short msgID) { int siz = 0; u_short mid = 0; @@ -145,14 +141,15 @@ mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_s return NULL; /* send subscribe */ - siz = mqtt_msgSUBSCRIBE(cli->buf, Topics, msgID, Dup, QoS); - if (siz == -1) + cli->buf = mqtt_msgSUBSCRIBE(Topics, msgID); + if (!cli->buf) return NULL; - 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; return NULL; - } + } else + mqtt_msgFree(&cli->buf, 0); if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { return NULL; @@ -170,13 +167,14 @@ mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_s return NULL; if (msgID != mid) { free(qoses); - mqtt_SetErr(EBADMSG, "Receive different message ID %hu != %hu", msgID, mid); + mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, mid); return NULL; } return qoses; } +#if 0 /* * mqtt_cli_Unsubscribe() - Unsubscribe from broker * @@ -192,7 +190,6 @@ mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt u_short msgID, u_char Dup, u_char QoS) { int siz = 0; - u_short mid = 0; if (!cli || !Topics) return -1; @@ -205,7 +202,8 @@ mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt if (siz == -1) { LOGERR; return -1; - } + } else + memset(cli->buf->msg_base, 0, cli->buf->msg_len); if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { return -1; @@ -222,9 +220,123 @@ mqtt_cli_Unsubscribe(mqtt_cli_t * __restrict cli, mqtt if (siz == -1) return -1; if (msgID != siz) { - mqtt_SetErr(EBADMSG, "Receive different message ID %hu != %hu", msgID, mid); + mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); return -1; } return 0; } + +/* + * mqtt_cli_Publish() - Publish message to broker + * + * @cli = connected client + * @msgID = Message ID + * @Dup = Duplicated request + * @QoS = Message QoS + * @Retain = Retain message + * @csTopic = Topic + * @pData = Data + * @datLen = Data length + * return: -1 error or > -1 sended bytes + */ +int +mqtt_cli_Publish(mqtt_cli_t * __restrict cli, u_short msgID, u_char Dup, u_char QoS, u_char Retain, + const char *csTopic, const void *pData, int datLen) +{ + int wlen = 0, siz = 0; + + if (!cli || !csTopic) + return -1; + + /* send publish */ + siz = mqtt_msgPUBLISH(cli->buf, csTopic, msgID, Dup, QoS, Retain, pData, datLen); + if (siz == -1) + return -1; + siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL); + if (siz == -1) { + LOGERR; + return -1; + } else { + wlen = siz; + memset(cli->buf->msg_base, 0, cli->buf->msg_len); + } + + if (QoS == MQTT_QOS_ONCE) /* no reply */ + goto end; + + if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { + return -1; + } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1)) + return -1; + + /* receive PUBxxx */ + siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); + if (siz == -1) { + LOGERR; + return -1; + } + + if (QoS == MQTT_QOS_ACK) { /* reply with PUBACK */ + siz = mqtt_readPUBACK(cli->buf); + if (siz == -1) + return -1; + if (msgID != siz) { + mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); + return -1; + } + goto end; + } else { /* reply with PUBREC */ + siz = mqtt_readPUBREC(cli->buf); + if (siz == -1) + return -1; + if (msgID != siz) { + mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); + return -1; + } + } + + do { + /* send publish release QoS == 2 */ + siz = mqtt_msgPUBREL(cli->buf, msgID); + if (siz == -1) + return -1; + siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL); + if (siz == -1) { + LOGERR; + return -1; + } else + memset(cli->buf->msg_base, 0, cli->buf->msg_len); + + if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { + return -1; + } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1)) { + if (Dup++ > 1) + return -1; + else + continue; + } + + /* receive PUBCOMP */ + siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); + if (siz == -1) { + LOGERR; + return -1; + } + + siz = mqtt_readPUBCOMP(cli->buf); + if (siz == -1) + return -1; + if (msgID != siz) { + mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); + if (Dup++ > 1) + return -1; + else + continue; + } + } while (0); + +end: + return wlen; +} +#endif