--- mqtt/src/mqtt_pub.c 2011/12/24 17:30:29 1.1.2.3 +++ mqtt/src/mqtt_pub.c 2012/01/05 10:26:02 1.1.2.6 @@ -19,6 +19,7 @@ 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" @@ -49,7 +50,44 @@ cleanArgs(struct tagArgs * __restrict args) static int Publish(int sock) { - return 0; + int siz = 0; + + siz = mqtt_msgPUBLISH(args->msg, AIT_GET_STR(&args->Publish), args->MsgID, args->Dup, + args->QoS, args->Retain, AIT_GET_PTR2(&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; + + if ((siz = RecvFrom(sock)) == -1 || !siz) + return -1; + /* QoS == MQTT_QOS_ACK, wait for PUBACK */ + if (args->QoS == MQTT_QOS_ACK) + return mqtt_readPUBACK(args->msg); + + + /* QoS == MQTT_QOS_EXACTLY */ + if (mqtt_readPUBREC(args->msg) != args->MsgID) { + printf("Error:: Message not delivered\n"); + return -1; + } + + siz = mqtt_msgPUBREL(args->msg, args->MsgID); + if (siz == -1) { + printf("Error:: msgPUBREL #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); + return -1; + } + if (SendTo(sock, siz) == -1) + return -1; + + if ((siz = RecvFrom(sock)) == -1 || !siz) + return -1; + return mqtt_readPUBCOMP(args->msg); } @@ -80,7 +118,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:fvh")) != -1) + while ((ch = getopt(argc, argv, "T:U:P:p:q:drCW:M:m:fvh")) != -1) switch (ch) { case 'T': args->ka = (u_short) strtol(optarg, NULL, 0); @@ -101,6 +139,9 @@ 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; @@ -158,6 +199,8 @@ main(int argc, char **argv) Usage(); return 1; } + if (args->QoS && !args->MsgID) + args->MsgID = MQTT_DEFAULT_MSGID; ioVERBOSE(1) printf("Connecting to %s:%d ...\n", io_n2addr(&args->addr, &val), io_n2port(&args->addr)); if ((sock = InitClient()) == -1) { @@ -166,6 +209,12 @@ main(int argc, char **argv) return 2; } + if (args->isFile && !OpenFile()) { + args->free(args); + free(args); + return 3; + } + printf("Connected ... "); switch ((ret = ConnectClient(sock))) { case -1: @@ -192,14 +241,14 @@ main(int argc, char **argv) } if (ret == MQTT_RETCODE_ACCEPTED) { - ret = Publish(sock); - shutdown(sock, SHUT_RDWR); + ret = !(Publish(sock) == args->MsgID); CloseClient(sock); } else { close(sock); - ret = 3; + ret = 4; } + CloseFile(); args->free(args); free(args); return ret;