--- mqtt/src/mqtt_subs.c 2011/12/20 16:04:34 1.1 +++ mqtt/src/mqtt_subs.c 2012/01/27 15:05:38 1.2 @@ -0,0 +1,237 @@ +#include "global.h" +#include "rtlm.h" +#include "mqtt.h" +#include "client.h" + + +io_enableDEBUG; + +extern char compiled[], compiledby[], compilehost[]; +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-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" + "\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); + 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); + io_freeVars(&args->Subscribes); +} + +static int +Subscribe(int sock, FILE *lf) +{ + return 0; +} + + +int +main(int argc, char **argv) +{ + char ch, batch = 1; + ait_val_t *v, val; + u_short port = atoi(MQTT_PORT); + int sock, ret = 0; + char 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->Subscribes = io_allocVars(1))) { + printf("Error:: in subscribes array #%d - %s\n", io_GetErrno(), io_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:CDvh")) != -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': + v = io_allocVar(); + if (!v) { + printf("Error:: not enough memory #%d - %s\n", errno, strerror(errno)); + args->free(args); + free(args); + return 1; + } else + AIT_SET_STR(v, optarg); + io_arrayElem(args->Subscribes, io_arraySize(args->Subscribes), v); + 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 'h': + default: + args->free(args); + free(args); + Usage(); + return 1; + } + argc -= optind; + argv += optind; + if (argc < 3) { + 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]); + AIT_FREE_VAL(&args->Publish); + AIT_SET_STR(&args->Publish, argv[2]); + } + if (argc > 3) { + AIT_FREE_VAL(&args->Value); + AIT_SET_STR(&args->Value, argv[3]); + } + 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; + } + ioVERBOSE(1) printf("Connecting to %s:%d ...\n", io_n2addr(&args->addr, &val), io_n2port(&args->addr)); + + if ((sock = InitClient()) == -1) { + args->free(args); + free(args); + return 2; + } + + printf("Connected ... "); + switch ((ret = ConnectClient(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 = Subscribe(sock, lf); + fclose(lf); + shutdown(sock, SHUT_RDWR); + } else + printf("Error:: in subscribe file #%d - %s\n", errno, strerror(errno)); + CloseClient(sock); + } else { + close(sock); + ret = 3; + } + + args->free(args); + free(args); + return ret; +}