--- libaitmqtt/src/cliside.c 2012/04/07 21:02:48 1.1.2.2 +++ libaitmqtt/src/cliside.c 2012/06/20 15:02:24 1.2 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: cliside.c,v 1.1.2.2 2012/04/07 21:02:48 misho Exp $ +* $Id: cliside.c,v 1.2 2012/06/20 15:02:24 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -46,3 +46,288 @@ 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; +}