--- libaitmqtt/src/cliside.c 2012/05/01 01:12:16 1.1.2.3 +++ 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.3 2012/05/01 01:12:16 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 @@ -50,10 +50,11 @@ SUCH DAMAGE. * 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) +mqtt_cli_Open(struct sockaddr *addr, u_short timeout) { mqtt_cli_t *cli; @@ -67,6 +68,7 @@ mqtt_cli_Open(struct sockaddr *addr) } 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; @@ -118,4 +120,59 @@ mqtt_cli_Close(mqtt_cli_t ** __restrict cli) 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; }