--- libaitmqtt/src/cliside.c 2012/01/27 08:28:09 1.1 +++ libaitmqtt/src/cliside.c 2012/06/20 15:02:24 1.2 @@ -0,0 +1,333 @@ +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: cliside.c,v 1.2 2012/06/20 15:02:24 misho Exp $ +* +************************************************************************** +The ELWIX and AITNET software is distributed under the following +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 + by Michael Pounov . All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions +are met: +1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. +2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. +3. All advertising materials mentioning features or use of this software + must display the following acknowledgement: +This product includes software developed by Michael Pounov +ELWIX - Embedded LightWeight unIX and its contributors. +4. Neither the name of AITNET nor the names of its contributors + may be used to endorse or promote products derived from this software + without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS +OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) +HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF +SUCH DAMAGE. +*/ +#include "global.h" + + +/* + * mqtt_cli_Open() - Open client connection to MQTT broker + * + * @addr = brokers address + * @timeout = timeout + * return: NULL error or !=NULL connected to broker + */ +mqtt_cli_t * +mqtt_cli_Open(struct sockaddr *addr, u_short timeout) +{ + mqtt_cli_t *cli; + + if (!addr) + return NULL; + + cli = malloc(sizeof(mqtt_cli_t)); + if (!cli) { + LOGERR; + return NULL; + } else + memset(cli, 0, sizeof(mqtt_cli_t)); + + cli->timeout = timeout; + cli->sock = socket(addr->sa_family, SOCK_STREAM, IPPROTO_TCP); + if (cli->sock == -1) { + LOGERR; + free(cli); + return NULL; + } + if (connect(cli->sock, addr, addr->sa_len) == -1) { + LOGERR; + close(cli->sock); + free(cli); + return NULL; + } + + cli->buf = mqtt_msgAlloc(USHRT_MAX); + if (!cli->buf) { + close(cli->sock); + free(cli); + return NULL; + } + + return cli; +} + +/* + * mqtt_cli_Close() - Close client connection + * + * @cli = connected client + * return: -1 error or 0 disconnected client and freed all resources + */ +int +mqtt_cli_Close(mqtt_cli_t ** __restrict cli) +{ + int siz = 0; + + 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); + if (siz > -1) + shutdown((*cli)->sock, SHUT_RDWR); + } + close((*cli)->sock); + + mqtt_msgFree(&(*cli)->buf, 42); + + free(*cli); + *cli = NULL; + return 0; +} + +/* + * mqtt_cli_Subscribe() - Subscribe to broker + * + * @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! + */ +u_char * +mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t * __restrict Topics, + u_short msgID, u_char Dup, u_char QoS) +{ + int siz = 0; + u_short mid = 0; + u_char *qoses = NULL; + + if (!cli || !Topics) + return NULL; + + /* send subscribe */ + siz = mqtt_msgSUBSCRIBE(cli->buf, Topics, msgID, Dup, QoS); + if (siz == -1) + return NULL; + siz = send(cli->sock, cli->buf->msg_base, siz, MSG_NOSIGNAL); + if (siz == -1) { + LOGERR; + return NULL; + } + + if ((siz = mqtt_wait4data(cli->sock, cli->timeout, POLLIN | POLLPRI)) == -1) { + return NULL; + } else if (siz && mqtt_KeepAlive(cli->sock, cli->timeout, 1)) + return NULL; + + /* receive suback */ + siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); + if (siz == -1) { + LOGERR; + return NULL; + } + siz = mqtt_readSUBACK(cli->buf, &mid, &qoses); + if (siz == -1) + return NULL; + if (msgID != mid) { + free(qoses); + 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; + } + + 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; + + 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; + } + + 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 = 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; +}