--- mqtt/src/mqtt_pub.c 2012/04/11 15:08:27 1.2.2.2 +++ mqtt/src/mqtt_pub.c 2012/07/03 12:46:00 1.4 @@ -1,3 +1,48 @@ +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: mqtt_pub.c,v 1.4 2012/07/03 12:46:00 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" #include "rtlm.h" #include "mqtt.h" @@ -19,7 +64,6 @@ Usage(void) " Syntax: mqtt_pub [options] \n\n" "\t-f\t\t\t'value_for_publish' is file name instead text\n" "\t-q \t\tQoS level (0-at most 1, 1-at least 1, 2-exactly 1)\n" - "\t-m \t\tMessage ID for publish message\n" "\t-d\t\t\tSend duplicate message\n" "\t-r\t\t\tRetain message from broker\n\n" "\t-C\t\t\tNot clear before connect!!!\n" @@ -51,66 +95,25 @@ static int Publish(int sock) { int siz = 0; + u_short mid = 0; - siz = mqtt_msgPUBLISH(args->msg, AIT_GET_STR(&args->Publish), args->MsgID, args->Dup, - args->QoS, args->Retain, AIT_ADDR(&args->Value), AIT_LEN(&args->Value)); - if (siz == -1) { - printf("Error:: msgPUBLISH #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); - return -1; - } - if (SendTo(sock, siz) == -1) - return -1; - /* QoS == MQTT_QOS_ONCE, no wait for reply */ - if (args->QoS == MQTT_QOS_ONCE) - return 0; +#ifdef __NetBSD__ + srandom(getpid() ^ time(NULL)); +#else + srandomdev(); +#endif + mid = random() % USHRT_MAX; - if ((siz = RecvFrom(sock)) == -1 || !siz) - return -1; - /* QoS == MQTT_QOS_ACK, wait for PUBACK */ - if (args->QoS == MQTT_QOS_ACK) { - siz = mqtt_readPUBACK(args->msg); - if (siz == args->MsgID) - return siz; - if (!args->Dup) { - args->Dup++; - return Publish(sock); - } - goto end; - } - - - /* QoS == MQTT_QOS_EXACTLY */ - siz = mqtt_readPUBREC(args->msg); - if (siz != args->MsgID) { - if (!args->Dup) { - args->Dup++; - return Publish(sock); - } - goto end; - } - - siz = mqtt_msgPUBREL(args->msg, args->MsgID); + printf(" > Execute PUBLISH request #%d ... ", mid); + siz = mqtt_cli_Publish(args->cli, mid, args->Dup, args->QoS, args->Retain, + AIT_GET_STR(&args->Publish), AIT_ADDR(&args->Value), AIT_LEN(&args->Value)); if (siz == -1) { - printf("Error:: msgPUBREL #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); + printf("Error:: Publish #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); return -1; - } - if (SendTo(sock, siz) == -1) - return -1; + } else + printf("Sended %d bytes \n", siz); - if ((siz = RecvFrom(sock)) == -1 || !siz) - return -1; - - siz = mqtt_readPUBCOMP(args->msg); - if (siz == args->MsgID) - return siz; - if (!args->Dup) { - args->Dup++; - return Publish(sock); - } - -end: - printf("Error:: Message not delivered\n"); - return -1; + return siz; } @@ -120,9 +123,9 @@ main(int argc, char **argv) char ch; ait_val_t val; u_short port = atoi(MQTT_PORT); - int sock, ret = 0; + int ret = 0; - if (!(args = malloc(sizeof(struct tagArgs)))) { + if (!(args = io_malloc(sizeof(struct tagArgs)))) { printf("Error:: in alloc arguments #%d - %s\n", errno, strerror(errno)); return 1; } else @@ -132,7 +135,7 @@ main(int argc, char **argv) if (!(args->msg = mqtt_msgAlloc(USHRT_MAX))) { printf("Error:: in mqtt buffer #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); args->free(args); - free(args); + io_free(args); return 1; } @@ -141,7 +144,7 @@ main(int argc, char **argv) AIT_SET_STR(&args->Pass, ""); args->ka = MQTT_KEEPALIVE; - while ((ch = getopt(argc, argv, "T:U:P:p:q:drCW:M:m:fvh")) != -1) + while ((ch = getopt(argc, argv, "T:U:P:p:q:drCW:M:fvh")) != -1) switch (ch) { case 'T': args->ka = (u_short) strtol(optarg, NULL, 0); @@ -162,9 +165,6 @@ main(int argc, char **argv) AIT_FREE_VAL(&args->Pass); AIT_SET_STR(&args->Pass, optarg); break; - case 'm': - args->MsgID = (u_short) strtol(optarg, NULL, 0); - break; case 'p': port = (u_short) strtol(optarg, NULL, 0); break; @@ -173,7 +173,7 @@ main(int argc, char **argv) if (args->QoS > MQTT_QOS_EXACTLY) { printf("Error:: invalid QoS level %d\n", args->QoS); args->free(args); - free(args); + io_free(args); return 1; } break; @@ -195,7 +195,7 @@ main(int argc, char **argv) case 'h': default: args->free(args); - free(args); + io_free(args); Usage(); return 1; } @@ -204,7 +204,7 @@ main(int argc, char **argv) if (argc < 4) { printf("Error:: host for connect not found, connection id, topic or value not supplied!\n\n"); args->free(args); - free(args); + io_free(args); Usage(); return 1; } else { @@ -218,28 +218,27 @@ main(int argc, char **argv) if (!io_gethostbyname(*argv, port, &args->addr)) { printf("Error:: host not valid #%d - %s\n", io_GetErrno(), io_GetError()); args->free(args); - free(args); + io_free(args); Usage(); return 1; } - if (args->QoS && !args->MsgID) - args->MsgID = MQTT_DEFAULT_MSGID; printf("Connecting to %s:%d ... ", io_n2addr(&args->addr, &val), io_n2port(&args->addr)); AIT_FREE_VAL(&val); - if ((sock = InitClient()) == -1) { + if (!(args->cli = mqtt_cli_Open(&args->addr.sa, args->ka))) { args->free(args); - free(args); + io_free(args); return 2; } if (args->isFile && !OpenFile()) { + mqtt_cli_Close(&args->cli); args->free(args); - free(args); + io_free(args); return 3; } - switch ((ret = ConnectClient(sock))) { + switch ((ret = ConnectClient(args->cli->sock))) { case -1: printf(">> FAILED!\n"); break; @@ -264,15 +263,14 @@ main(int argc, char **argv) } if (ret == MQTT_RETCODE_ACCEPTED) { - ret = !(Publish(sock) == args->MsgID); - CloseClient(sock); - } else { - close(sock); + ret = (Publish(args->cli->sock) == -1); + } else ret = 4; - } + mqtt_cli_Close(&args->cli); + CloseFile(); args->free(args); - free(args); + io_free(args); return ret; }