![]() ![]() | ![]() |
added SIGINT for mqttd check QoS into mqttd
1: #include "global.h" 2: #include "mqttd.h" 3: #include "rtlm.h" 4: #include "utils.h" 5: #include "daemon.h" 6: 7: 8: io_enableDEBUG; 9: 10: cfg_root_t cfg; 11: sessions_t Sessions; 12: sched_root_task_t *root; 13: sqlite3 *acc, *pub; 14: FILE *logg; 15: extern char compiled[], compiledby[], compilehost[]; 16: static char szCfgName[MAXPATHLEN]; 17: volatile intptr_t Kill; 18: 19: 20: static void 21: Usage(void) 22: { 23: printf( " -= MQTT Broker =- MQTT Service from ELWIX\n" 24: "=== %s@%s === Compiled: %s ===\n\n" 25: "\t-c <config>\tService config\n" 26: "\t-b\t\tBatch mode\n" 27: "\t-v\t\tVerbose (more -vvv, more verbose)\n" 28: "\t-h\t\tHelp! This screen\n\n", 29: compiledby, compilehost, compiled); 30: } 31: 32: static void 33: sigHand(int sig) 34: { 35: int stat; 36: 37: switch (sig) { 38: case SIGHUP: 39: cfgUnloadConfig(&cfg); 40: if (!cfgLoadConfig(szCfgName, &cfg)) { 41: ioDEBUG(1, "Config reload OK!"); 42: break; 43: } 44: 45: ioLIBERR(cfg); 46: case SIGINT: 47: case SIGTERM: 48: ioDEBUG(1, "Terminate MQTT service in progress"); 49: Kill++; 50: break; 51: case SIGCHLD: 52: while (waitpid(-1, &stat, WNOHANG) > 0); 53: break; 54: case SIGPIPE: 55: break; 56: } 57: } 58: 59: 60: int 61: main(int argc, char **argv) 62: { 63: char ch, batch = 0; 64: register int i; 65: int sock = -1, ret = 0; 66: struct passwd *pass; 67: struct sigaction sa; 68: ait_val_t v; 69: 70: TAILQ_INIT(&Sessions); 71: 72: strlcpy(szCfgName, DEFAULT_CONFIG, sizeof szCfgName); 73: while ((ch = getopt(argc, argv, "hvbc:")) != -1) 74: switch (ch) { 75: case 'c': 76: strlcpy(szCfgName, optarg, sizeof szCfgName); 77: break; 78: case 'b': 79: batch++; 80: break; 81: case 'v': 82: io_incDebug; 83: break; 84: case 'h': 85: default: 86: Usage(); 87: return 1; 88: } 89: argc -= optind; 90: argv += optind; 91: 92: if (cfgLoadConfig(szCfgName, &cfg)) { 93: printf("Error:: can't load #%d - %s\n", cfg_GetErrno(), cfg_GetError()); 94: return 1; 95: } 96: openlog("mqttd", LOG_PID | LOG_CONS, LOG_DAEMON); 97: /* load 3 plugins */ 98: for (i = 0; i < 3; i++) 99: if (!mqttLoadRTLM(&cfg, i)) { 100: printf("Error:: Can't load RTL module\n"); 101: mqttUnloadRTLM(acc); 102: mqttUnloadRTLM(pub); 103: mqttUnloadRTLM(logg); 104: cfgUnloadConfig(&cfg); 105: closelog(); 106: return 2; 107: } 108: acc = call.OpenACC(&cfg); 109: if (!acc) { 110: ret = 3; 111: goto end; 112: } 113: pub = call.OpenPUB(&cfg); 114: if (!pub) { 115: ret = 3; 116: goto end; 117: } 118: logg = call.OpenLOG(&cfg); 119: if (!logg) { 120: ret = 3; 121: goto end; 122: } 123: 124: 125: if (mqttMkDir(&cfg)) { 126: printf("Error:: in statedir #%d - %s\n", errno, strerror(errno)); 127: ret = 3; 128: goto end; 129: } 130: 131: if (!batch) 132: switch (fork()) { 133: case -1: 134: printf("Error:: in fork() #%d - %s\n", errno, strerror(errno)); 135: ret = 5; 136: goto end; 137: case 0: 138: setsid(); 139: 140: ret = open("/dev/null", O_RDWR); 141: if (ret != -1) { 142: dup2(ret, STDIN_FILENO); 143: dup2(ret, STDOUT_FILENO); 144: dup2(ret, STDERR_FILENO); 145: close(ret); 146: } 147: ioDEBUG(2, "Welcome MQTT service into shadow land!"); 148: break; 149: default: 150: ioDEBUG(2, "MQTT service go to shadow land ..."); 151: sleep(1); 152: ret = 0; 153: goto end; 154: } 155: else 156: ioDEBUG(1, "Start service in batch mode ..."); 157: 158: memset(&sa, 0, sizeof sa); 159: sigemptyset(&sa.sa_mask); 160: sa.sa_handler = sigHand; 161: sigaction(SIGHUP, &sa, NULL); 162: sigaction(SIGINT, &sa, NULL); 163: sigaction(SIGTERM, &sa, NULL); 164: sigaction(SIGCHLD, &sa, NULL); 165: sigaction(SIGPIPE, &sa, NULL); 166: ioDEBUG(2, "Service is ready for starting engine ..."); 167: 168: if ((sock = srv_Socket(&cfg)) == -1) { 169: ret = 4; 170: goto end; 171: } 172: 173: cfg_loadAttribute(&cfg, "mqttd", "user", &v, MQTT_USER); 174: pass = getpwnam(AIT_GET_STR(&v)); 175: AIT_FREE_VAL(&v); 176: if (pass) { 177: setgid(pass->pw_gid); 178: setuid(pass->pw_uid); 179: ioDEBUG(2, "Try to change group #%d and user #%d", pass->pw_gid, pass->pw_uid); 180: } 181: 182: if (!(root = schedBegin())) { 183: ioLIBERR(sched); 184: ret = 6; 185: goto end; 186: } 187: 188: /* go catch the cat ... */ 189: Run(sock); 190: 191: schedEnd(&root); 192: end: /* free all resources */ 193: srv_Close(sock); 194: call.CloseLOG(logg); 195: call.ClosePUB(pub); 196: call.CloseACC(acc); 197: mqttUnloadRTLM(acc); 198: mqttUnloadRTLM(pub); 199: mqttUnloadRTLM(logg); 200: cfgUnloadConfig(&cfg); 201: closelog(); 202: return ret; 203: }