version 1.1.2.2, 2012/04/07 21:02:48
|
version 1.3.12.4, 2022/09/16 04:18:17
|
Line 12 terms:
|
Line 12 terms:
|
All of the documentation and software included in the ELWIX and AITNET |
All of the documentation and software included in the ELWIX and AITNET |
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org> |
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org> |
|
|
Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 | Copyright 2004 - 2022 |
by Michael Pounov <misho@elwix.org>. All rights reserved. |
by Michael Pounov <misho@elwix.org>. All rights reserved. |
|
|
Redistribution and use in source and binary forms, with or without |
Redistribution and use in source and binary forms, with or without |
Line 46 SUCH DAMAGE.
|
Line 46 SUCH DAMAGE.
|
#include "global.h" |
#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 = e_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; |
|
e_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); |
|
e_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; |
|
|
|
if (!cli || !*cli) |
|
return -1; |
|
|
|
/* send disconnect */ |
|
(*cli)->buf = mqtt_msgDISCONNECT(); |
|
if ((*cli)->buf) { |
|
siz = send((*cli)->sock, (*cli)->buf->msg_base, (*cli)->buf->msg_len, MSG_NOSIGNAL); |
|
if (siz > -1) |
|
shutdown((*cli)->sock, SHUT_RDWR); |
|
} |
|
close((*cli)->sock); |
|
|
|
mqtt_msgFree(&(*cli)->buf, 0); |
|
|
|
e_free(*cli); |
|
*cli = NULL; |
|
return 0; |
|
} |
|
|
|
/* |
|
* mqtt_cli_Subscribe() - Subscribe to broker |
|
* |
|
* @cli = connected client |
|
* @Topics = Topics for subscribes |
|
* @msgID = Message ID |
|
* return: NULL error or !=NULL allocated array with subscribed QoS responses, |
|
* must be e_free() result! |
|
*/ |
|
u_char * |
|
mqtt_cli_Subscribe(mqtt_cli_t * __restrict cli, mqtt_subscr_t * __restrict Topics, u_short msgID) |
|
{ |
|
int siz = 0; |
|
u_short mid = 0; |
|
u_char *qoses = NULL; |
|
|
|
if (!cli || !Topics) |
|
return NULL; |
|
|
|
/* send subscribe */ |
|
cli->buf = mqtt_msgSUBSCRIBE(Topics, msgID); |
|
if (!cli->buf) |
|
return NULL; |
|
siz = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); |
|
if (siz == -1) { |
|
LOGERR; |
|
return NULL; |
|
} else |
|
mqtt_msgFree(&cli->buf, 0); |
|
|
|
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 */ |
|
cli->buf = mqtt_msgAlloc(BUFSIZ); |
|
if (!cli->buf) |
|
return NULL; |
|
siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); |
|
if (siz == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return NULL; |
|
} |
|
siz = mqtt_readSUBACK(cli->buf, &mid, &qoses); |
|
if (siz == -1) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
return NULL; |
|
} |
|
if (msgID != mid) { |
|
if (qoses) |
|
e_free(qoses); |
|
mqtt_msgFree(&cli->buf, 0); |
|
mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, mid); |
|
return NULL; |
|
} |
|
|
|
mqtt_msgFree(&cli->buf, 0); |
|
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 */ |
|
cli->buf = mqtt_msgUNSUBSCRIBE(Topics, msgID, Dup, QoS); |
|
if (!cli->buf) |
|
return -1; |
|
siz = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); |
|
if (siz == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} else |
|
mqtt_msgFree(&cli->buf, 0); |
|
|
|
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 */ |
|
cli->buf = mqtt_msgAlloc(BUFSIZ); |
|
if (!cli->buf) |
|
return -1; |
|
siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); |
|
if (siz == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
siz = mqtt_readUNSUBACK(cli->buf); |
|
if (siz == -1) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
if (msgID != siz) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); |
|
return -1; |
|
} |
|
|
|
mqtt_msgFree(&cli->buf, 0); |
|
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 */ |
|
cli->buf = mqtt_msgPUBLISH(csTopic, msgID, Dup, QoS, Retain, pData, datLen); |
|
if (!cli->buf) |
|
return -1; |
|
wlen = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); |
|
if (wlen == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} else |
|
mqtt_msgFree(&cli->buf, 0); |
|
|
|
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 */ |
|
cli->buf = mqtt_msgAlloc(BUFSIZ); |
|
if (!cli->buf) |
|
return -1; |
|
siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); |
|
if (siz == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
|
|
if (QoS == MQTT_QOS_ACK) { /* reply with PUBACK */ |
|
siz = mqtt_readPUBACK(cli->buf); |
|
if (siz == -1) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
if (msgID != siz) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); |
|
return -1; |
|
} |
|
mqtt_msgFree(&cli->buf, 0); |
|
goto end; |
|
} else { /* reply with PUBREC */ |
|
siz = mqtt_readPUBREC(cli->buf); |
|
if (siz == -1) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
if (msgID != siz) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); |
|
return -1; |
|
} |
|
mqtt_msgFree(&cli->buf, 0); |
|
} |
|
|
|
do { |
|
/* send publish release QoS == 2 */ |
|
cli->buf = mqtt_msgPUBREL(msgID); |
|
if (!cli->buf) |
|
return -1; |
|
siz = send(cli->sock, cli->buf->msg_base, cli->buf->msg_len, MSG_NOSIGNAL); |
|
if (siz == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} else |
|
mqtt_msgFree(&cli->buf, 0); |
|
|
|
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 */ |
|
cli->buf = mqtt_msgAlloc(BUFSIZ); |
|
if (!cli->buf) |
|
return -1; |
|
siz = recv(cli->sock, cli->buf->msg_base, cli->buf->msg_len, 0); |
|
if (siz == -1) { |
|
LOGERR; |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
|
|
siz = mqtt_readPUBCOMP(cli->buf); |
|
if (siz == -1) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
return -1; |
|
} |
|
if (msgID != siz) { |
|
mqtt_msgFree(&cli->buf, 0); |
|
mqtt_SetErr(ECANCELED, "Receive different message ID %hu != %hu", msgID, siz); |
|
if (Dup++ > 1) |
|
return -1; |
|
else |
|
continue; |
|
} |
|
mqtt_msgFree(&cli->buf, 0); |
|
} while (0); |
|
|
|
end: |
|
return wlen; |
|
} |