--- mqtt/src/mqtt_subs.c 2012/05/01 22:30:31 1.2.2.5 +++ mqtt/src/mqtt_subs.c 2012/06/01 13:13:40 1.2.2.11 @@ -19,8 +19,8 @@ Usage(void) "=== %s@%s === Compiled: %s ===\n\n" " Syntax: mqtt_subs [options] [exec_script ]\n\n" "\t-l \t\tSave received values to file\n" + "\t-u\t\t\tUnsubscribe given topic(s)\n" "\t-s \tSubscribe for this topic, if wish add different |QoS to topic\n" - "\t-q \t\tQoS level (0-at most 1, 1-at least 1, 2-exactly 1)\n" "\t-d\t\t\tSend duplicate message\n\n" "\t-C\t\t\tNot clear before connect!\n" "\t-p \t\tDifferent port for connect (default: 1883)\n" @@ -52,44 +52,60 @@ cleanArgs(struct tagArgs * __restrict args) static int Subscribe(int sock, FILE *lf) { - int siz; - u_char *qoses; - u_short mid[2]; + u_char *qoses, *qos; + u_short mid; + mqtt_subscr_t *sub; +#ifdef __NetBSD__ + srandom(getpid() ^ time(NULL)); +#else srandomdev(); - mid[0] = random() % USHRT_MAX; +#endif + mid = random() % USHRT_MAX; - siz = mqtt_msgSUBSCRIBE(args->msg, args->subscr, mid[0], args->Dup, args->QoS); - if (siz == -1) { - printf("Error:: in msgSUBSCRIBE #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); + printf(" > Execute SUBSCRIBE request #%d ... ", mid); + qoses = mqtt_cli_Subscribe(args->cli, args->subscr, mid, args->Dup, MQTT_QOS_ACK); + if (!qoses) { + printf("Error:: Subscribe #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); return -1; - } - siz = SendTo(sock, siz); - if (siz == -1) - return -1; + } else + printf("OK\n"); - siz = RecvFrom(sock); - if (siz == -1) - return -1; - siz = mqtt_readSUBACK(args->msg, &mid[1], &qoses); - if (siz == -1) { - printf("Error:: in readSUBACK #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); - return -1; - } - if (mid[1] != mid[0]) { - printf("Error:: received different connection ID %d != %d\n", mid[1], mid[0]); - return -1; - } + for (sub = args->subscr, qos = qoses; sub->sub_topic.msg_base; sub++, qos++) + printf(" + Topic %s with QoS %d subscribed %s\n", (char*) + sub->sub_topic.msg_base, sub->sub_ret, *qos ? "done" : "failed"); free(qoses); return 0; } +static int +Unsubscribe(int sock) +{ + u_short mid; +#ifdef __NetBSD__ + srandom(getpid() ^ time(NULL)); +#else + srandomdev(); +#endif + mid = random() % USHRT_MAX; + + printf(" > Execute UNSUBSCRIBE request #%d ... ", mid); + if (mqtt_cli_Unsubscribe(args->cli, args->subscr, mid, args->Dup, MQTT_QOS_ACK)) { + printf("Error:: Unsubscribe #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); + return -1; + } else + printf("OK\n"); + + return 0; +} + + int main(int argc, char **argv) { - char ch, idx = 0, batch = 1; + char ch, un = 0, idx = 0, batch = 1; ait_val_t val; u_short port = atoi(MQTT_PORT); mqtt_subscr_t *sub; @@ -97,14 +113,14 @@ main(int argc, char **argv) char *str, szStr[STRSIZ], szLogName[MAXPATHLEN] = { 0 }; FILE *lf; - if (!(args = malloc(sizeof(struct tagArgs)))) { + if (!(args = io_malloc(sizeof(struct tagArgs)))) { printf("Error:: in arguments #%d - %s\n", errno, strerror(errno)); return 1; } else memset(args, 0, sizeof(struct tagArgs)); if (!(args->subscr = mqtt_subAlloc(idx))) { printf("Error:: in subscribes array #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); - free(args); + io_free(args); return 1; } else args->free = cleanArgs; @@ -112,7 +128,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; } @@ -121,7 +137,7 @@ main(int argc, char **argv) AIT_SET_STR(&args->Pass, ""); args->ka = MQTT_KEEPALIVE; - while ((ch = getopt(argc, argv, "T:U:P:p:s:q:dl:W:M:CDvh")) != -1) + while ((ch = getopt(argc, argv, "T:U:P:p:s:q:dl:W:M:CDvuh")) != -1) switch (ch) { case 'T': args->ka = (u_short) strtol(optarg, NULL, 0); @@ -150,7 +166,7 @@ main(int argc, char **argv) if (!sub) { printf("Error:: #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); args->free(args); - free(args); + io_free(args); return 1; } else sub += idx++; @@ -173,7 +189,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; @@ -192,10 +208,13 @@ main(int argc, char **argv) case 'v': io_incDebug; break; + case 'u': + un = 1; + break; case 'h': default: args->free(args); - free(args); + io_free(args); Usage(); return 1; } @@ -204,7 +223,7 @@ main(int argc, char **argv) if (argc < 2) { printf("Error:: host for connect not found, connection id or topic not supplied!\n\n"); args->free(args); - free(args); + io_free(args); Usage(); return 1; } else { @@ -218,16 +237,16 @@ 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; } printf("Connecting to %s:%d ... ", io_n2addr(&args->addr, &val), io_n2port(&args->addr)); AIT_FREE_VAL(&val); - if (!(args->cli = mqtt_cli_Open(&args->addr.sa))) { + if (!(args->cli = mqtt_cli_Open(&args->addr.sa, args->ka))) { args->free(args); - free(args); + io_free(args); return 2; } @@ -262,6 +281,8 @@ main(int argc, char **argv) lf = stdout; if (lf) { ret = Subscribe(args->cli->sock, lf); + if (un) + Unsubscribe(args->cli->sock); fclose(lf); } else printf("Error:: in subscribe file #%d - %s\n", errno, strerror(errno)); @@ -271,6 +292,6 @@ main(int argc, char **argv) mqtt_cli_Close(&args->cli); args->free(args); - free(args); + io_free(args); return ret; }