--- libaitmqtt/src/cliside.c 2012/05/08 09:07:31 1.1.2.4 +++ libaitmqtt/src/cliside.c 2016/09/14 15:52:36 1.3.12.1 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: cliside.c,v 1.1.2.4 2012/05/08 09:07:31 misho Exp $ +* $Id: cliside.c,v 1.3.12.1 2016/09/14 15:52:36 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 - 2016 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -75,7 +75,13 @@ mqtt_cli_Open(struct sockaddr *addr, u_short timeout) 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); @@ -141,7 +147,7 @@ mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_s u_short mid = 0; u_char *qoses = NULL; - if (!cli) + if (!cli || !Topics) return NULL; /* send subscribe */ @@ -152,7 +158,8 @@ mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_s if (siz == -1) { LOGERR; return NULL; - } + } else + memset(cli->buf->msg_base, 0, cli->buf->msg_len); if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { return NULL; @@ -170,9 +177,174 @@ 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; +} + +/* + * mqtt_cli_Unsubscribe() - Unsubscribe from broker + * + * @cli = connected client + * @Topics = Topics for unsubscribes + * @msgID = Message ID + * @Dup = Duplicated request + * @QoS = Message QoS + * return: -1 error or 0 ok + */ +int +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; + + if (!cli || !Topics) + return -1; + + /* send unsubscribe */ + siz = mqtt_msgUNSUBSCRIBE(cli->buf, Topics, msgID, Dup, QoS); + 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)) + return -1; + + /* receive unsuback */ + siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); + if (siz == -1) { + LOGERR; + return -1; + } + siz = mqtt_readUNSUBACK(cli->buf); + if (siz == -1) + return -1; + if (msgID != siz) { + 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; }