--- libaitmqtt/src/cliside.c 2012/04/07 21:02:48 1.1.2.2 +++ libaitmqtt/src/cliside.c 2012/05/08 09:07:31 1.1.2.4 @@ -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.1.2.4 2012/05/08 09:07:31 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -46,3 +46,133 @@ 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) + 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(EBADMSG, "Receive different message ID %hu != %hu", msgID, mid); + return NULL; + } + + return qoses; +}