#include "global.h" #include "rtlm.h" #include "mqtt.h" #include "client.h" io_enableDEBUG; extern char compiled[], compiledby[], compilehost[]; struct tagArgs *args; static void Usage(void) { printf( " -= MQTT Publisher Client =- Publisher from ELWIX\n" "=== %s@%s === Compiled: %s ===\n\n" " 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" "\t-p \t\tDifferent port for connect (default: 1883)\n" "\t-T \t\tKeep alive timeout in seconds (default: 10sec)\n" "\t-U \t\tUsername\n" "\t-P \t\tPassword\n" "\t-W \t\tWill Topic\n" "\t-M \t\tWill Message\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); } static int Publish(int sock) { 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) { 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); 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; 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; } int main(int argc, char **argv) { char ch; ait_val_t val; u_short port = atoi(MQTT_PORT); int sock, ret = 0; if (!(args = malloc(sizeof(struct tagArgs)))) { printf("Error:: in alloc arguments #%d - %s\n", errno, strerror(errno)); return 1; } else memset(args, 0, sizeof(struct tagArgs)); 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:q:drCW:M:m:fvh")) != -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 'm': args->MsgID = (u_short) strtol(optarg, NULL, 0); break; case 'p': port = (u_short) strtol(optarg, NULL, 0); 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 'r': args->Retain++; break; case 'C': args->notClear++; break; case 'f': args->isFile++; break; case 'v': io_incDebug; break; case 'h': default: args->free(args); free(args); Usage(); return 1; } argc -= optind; argv += optind; if (argc < 4) { printf("Error:: host for connect not found, connection id, topic or value 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]); 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; } 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) { args->free(args); free(args); return 2; } if (args->isFile && !OpenFile()) { args->free(args); free(args); return 3; } 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) { ret = !(Publish(sock) == args->MsgID); CloseClient(sock); } else { close(sock); ret = 4; } CloseFile(); args->free(args); free(args); return ret; }