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