#include "global.h" #include "rtlm.h" #include "mqtt.h" #include "client.h" io_enableDEBUG; extern char compiled[], compiledby[], compilehost[]; volatile intptr_t Kill; struct tagArgs *args; static void Usage(void) { printf( " -= MQTT Subscriber Client =- Subscriber from ELWIX\n" "=== %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\tUnsubscribe given topic(s)\n" "\t-s \tSubscribe for this topic, if wish add different |QoS to topic\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" "\t-T \t\tKeep alive timeout in seconds\n" "\t-U \t\tUsername\n" "\t-P \t\tPassword\n" "\t-W \t\tWill Topic\n" "\t-M \t\tWill Message\n\n" "\t-D\t\t\tDaemon mode\n" "\t-v\t\t\tVerbose (more -vvv, more verbose)\n" "\t-h\t\t\tHelp! This screen\n\n", compiledby, compilehost, compiled); } static void cleanArgs(struct tagArgs * __restrict args) { mqtt_msgFree(&args->msg, 42); mqtt_subFree(&args->subscr); AIT_FREE_VAL(&args->Will.Msg); AIT_FREE_VAL(&args->Will.Topic); AIT_FREE_VAL(&args->User); AIT_FREE_VAL(&args->Pass); AIT_FREE_VAL(&args->Publish); AIT_FREE_VAL(&args->Value); AIT_FREE_VAL(&args->ConnID); } static int Subscribe(int sock, FILE *lf) { u_char *qoses, *qos; u_short mid; mqtt_subscr_t *sub; srandomdev(); mid = random() % USHRT_MAX; 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; } else printf("OK\n"); 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; srandomdev(); mid = random() % USHRT_MAX; 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; } return 0; } int main(int argc, char **argv) { char ch, un = 0, idx = 0, batch = 1; ait_val_t val; u_short port = atoi(MQTT_PORT); mqtt_subscr_t *sub; int ret = 0; char *str, szStr[STRSIZ], szLogName[MAXPATHLEN] = { 0 }; FILE *lf; if (!(args = 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); return 1; } else args->free = cleanArgs; if (!(args->msg = mqtt_msgAlloc(USHRT_MAX))) { printf("Error:: in mqtt buffer #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); args->free(args); free(args); return 1; } AIT_SET_STR(&args->ConnID, ""); AIT_SET_STR(&args->User, ""); AIT_SET_STR(&args->Pass, ""); args->ka = MQTT_KEEPALIVE; 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); break; case 'M': AIT_FREE_VAL(&args->Will.Msg); AIT_SET_STR(&args->Will.Msg, optarg); break; case 'W': AIT_FREE_VAL(&args->Will.Topic); AIT_SET_STR(&args->Will.Topic, optarg); break; case 'U': AIT_FREE_VAL(&args->User); AIT_SET_STR(&args->User, optarg); break; case 'P': AIT_FREE_VAL(&args->Pass); AIT_SET_STR(&args->Pass, optarg); break; case 'p': port = (u_short) strtol(optarg, NULL, 0); break; case 's': sub = mqtt_subRealloc(&args->subscr, idx + 1); if (!sub) { printf("Error:: #%d - %s\n", mqtt_GetErrno(), mqtt_GetError()); args->free(args); free(args); return 1; } else sub += idx++; strlcpy(szStr, optarg, sizeof szStr); if ((str = strchr(szStr, '|'))) { *str++ = 0; *str -= 0x30; if (*str < 0 || *str > MQTT_QOS_RESERVED) sub->sub_ret = (u_char) args->QoS; else sub->sub_ret = (u_char) *str; } else sub->sub_ret = (u_char) args->QoS; sub->sub_topic.msg_base = strdup(szStr); sub->sub_topic.msg_len = strlen(szStr); break; case 'q': args->QoS = (char) strtol(optarg, NULL, 0); if (args->QoS > MQTT_QOS_EXACTLY) { printf("Error:: invalid QoS level %d\n", args->QoS); args->free(args); free(args); return 1; } break; case 'd': args->Dup++; break; case 'C': args->notClear++; break; case 'l': strlcpy(szLogName, optarg, sizeof szLogName); break; case 'D': batch = 0; break; case 'v': io_incDebug; break; case 'u': un = 1; break; case 'h': default: args->free(args); free(args); Usage(); return 1; } argc -= optind; argv += optind; if (argc < 2) { printf("Error:: host for connect not found, connection id or topic not supplied!\n\n"); args->free(args); free(args); Usage(); return 1; } else { AIT_FREE_VAL(&args->ConnID); AIT_SET_STR(&args->ConnID, argv[1]); } if (argc > 2) { AIT_FREE_VAL(&args->Value); AIT_SET_STR(&args->Value, argv[2]); } if (!io_gethostbyname(*argv, port, &args->addr)) { printf("Error:: host not valid #%d - %s\n", io_GetErrno(), io_GetError()); args->free(args); 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, args->ka))) { args->free(args); free(args); return 2; } switch ((ret = ConnectClient(args->cli->sock))) { case -1: printf(">> FAILED!\n"); break; case MQTT_RETCODE_ACCEPTED: printf(">> OK\n"); break; case MQTT_RETCODE_REFUSE_VER: printf(">> Incorrect version\n"); break; case MQTT_RETCODE_REFUSE_ID: printf(">> Incorrect connectID\n"); break; case MQTT_RETCODE_REFUSE_UNAVAIL: printf(">> Service unavailable\n"); break; case MQTT_RETCODE_REFUSE_USERPASS: printf(">> Refuse user/pass\n"); break; case MQTT_RETCODE_DENIED: printf(">> DENIED.\n"); break; } if (ret == MQTT_RETCODE_ACCEPTED) { if (*szLogName) lf = fopen(szLogName, "w"); else lf = stdout; if (lf) { ret = un ? Unsubscribe(args->cli->sock) : Subscribe(args->cli->sock, lf); fclose(lf); } else printf("Error:: in subscribe file #%d - %s\n", errno, strerror(errno)); } else ret = 3; mqtt_cli_Close(&args->cli); args->free(args); free(args); return ret; }