Annotation of mqtt/src/daemon.c, revision 1.2.2.13
1.2 misho 1: #include "global.h"
2: #include "rtlm.h"
3: #include "utils.h"
4: #include "mqttd.h"
5: #include "mqttd_calls.h"
6:
7:
8: static void *startSession(sched_task_t *task);
1.2.2.10 misho 9: static pthread_attr_t attr;
1.2 misho 10:
11:
12: static inline struct tagSession *
13: initSession(int sock, ait_val_t * __restrict v)
14: {
15: struct tagSession *sess = NULL;
16: const char *str;
17:
18: ioTRACE(5);
19:
20: if (!v)
21: return NULL;
22:
23: sess = malloc(sizeof(struct tagSession));
24: if (!sess) {
1.2.2.9 misho 25: ioSYSERR(0);
1.2 misho 26: io_freeVar(v);
27: return NULL;
28: } else
29: memset(sess, 0, sizeof(struct tagSession));
30:
1.2.2.4 misho 31: pthread_mutex_init(&sess->sess_mtx, NULL);
32:
1.2.2.9 misho 33: SLIST_INIT(&sess->sess_subscr);
1.2 misho 34:
1.2.2.5 misho 35: str = cfg_getAttribute(&cfg, "mqttd", "retry");
1.2 misho 36: if (!str)
37: sess->sess_retry = DEFAULT_RETRY;
38: else
39: sess->sess_retry = strtol(str, NULL, 0);
40:
1.2.2.9 misho 41: if (!(sess->sess_root = schedBegin())) {
42: ioLIBERR(sched);
43: free(sess);
44: io_freeVar(v);
45: return NULL;
46: }
47:
1.2 misho 48: sess->sess_buf = mqtt_msgAlloc(USHRT_MAX);
49: if (!sess->sess_buf) {
1.2.2.9 misho 50: ioLIBERR(mqtt);
51: schedEnd(&sess->sess_root);
1.2 misho 52: free(sess);
53: io_freeVar(v);
54: return NULL;
55: }
56:
1.2.2.8 misho 57: /* init server actor */
1.2.2.2 misho 58: sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf);
59: if (!sess->sess_srv) {
60: ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
61: mqtt_msgFree(&sess->sess_buf, 42);
1.2.2.9 misho 62: schedEnd(&sess->sess_root);
1.2.2.2 misho 63: free(sess);
64: io_freeVar(v);
65: return NULL;
1.2.2.3 misho 66: } else {
67: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT);
68: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBLISH, cmdPUBLISH);
69: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PUBREL, cmdPUBREL);
70: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_SUBSCRIBE, cmdSUBSCRIBE);
71: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_UNSUBSCRIBE, cmdUNSUBSCRIBE);
72: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_PINGREQ, cmdPINGREQ);
73: mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_DISCONNECT, cmdDISCONNECT);
1.2.2.2 misho 74: }
75:
1.2 misho 76: sess->sess_sock = sock;
77: strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr);
78: io_freeVar(v);
79: return sess;
80: }
81:
82: static void
1.2.2.9 misho 83: finiSession(struct tagSession *sess)
1.2 misho 84: {
85: struct tagStore *store;
86:
87: ioTRACE(5);
88:
89: if (!sess)
90: return;
91:
92: if (call.FiniSessPUB)
93: call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
94:
1.2.2.6 misho 95: SESS_ELEM_LOCK(sess);
1.2 misho 96:
1.2.2.9 misho 97: while ((store = SLIST_FIRST(&sess->sess_subscr))) {
98: SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);
99:
100: if (store->st_subscr.sub_topic.msg_base)
101: free(store->st_subscr.sub_topic.msg_base);
102: if (store->st_subscr.sub_value.msg_base)
103: free(store->st_subscr.sub_value.msg_base);
104:
105: free(store);
106: }
1.2.2.6 misho 107: SESS_ELEM_UNLOCK(sess);
1.2.2.4 misho 108: pthread_mutex_destroy(&sess->sess_mtx);
1.2 misho 109:
1.2.2.9 misho 110: schedEnd(&sess->sess_root);
111:
1.2 misho 112: if (sess->sess_will.msg)
113: free(sess->sess_will.msg);
114: if (sess->sess_will.topic)
115: free(sess->sess_will.topic);
116:
1.2.2.9 misho 117: if (sess->sess_sock > STDERR_FILENO)
1.2 misho 118: srv_Close(sess->sess_sock);
119:
1.2.2.2 misho 120: mqtt_srv_Fini(&sess->sess_srv);
1.2 misho 121: mqtt_msgFree(&sess->sess_buf, 42);
122:
123: free(sess);
124: }
125:
126: static void
127: stopSession(struct tagSession *sess)
128: {
129: mqtt_msg_t msg = { NULL, 0 };
130: int ret;
131:
132: ioTRACE(4);
133:
134: assert(sess);
135:
1.2.2.6 misho 136: SESS_LOCK;
1.2 misho 137: TAILQ_REMOVE(&Sessions, sess, sess_node);
1.2.2.6 misho 138: SESS_UNLOCK;
1.2 misho 139:
140: ret = mqtt_msgDISCONNECT(&msg);
1.2.2.9 misho 141: send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL);
142: free(msg.msg_base);
1.2 misho 143:
144: ioDEBUG(1, "Close socket=%d", sess->sess_sock);
1.2.2.9 misho 145: finiSession(sess);
1.2 misho 146:
147: call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,
148: sess->sess_addr, sess->sess_user);
149: }
150:
151: static int
152: KASession(struct tagSession *sess)
153: {
154: mqtt_msg_t msg = { NULL, 0 };
155: int ret;
156: struct pollfd pfd;
157:
158: ioTRACE(4);
159:
160: assert(sess);
161:
162: /* ping request */
163: ret = mqtt_msgPINGREQ(&msg);
1.2.2.9 misho 164: if ((ret = send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) {
1.2 misho 165: ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
166: return -1;
167: } else {
168: ioDEBUG(5, "Sended %d bytes for ping request", ret);
169: free(msg.msg_base);
170: memset(&msg, 0, sizeof msg);
171: }
172:
173: pfd.fd = sess->sess_sock;
174: pfd.events = POLLIN | POLLPRI;
175: if ((ret = poll(&pfd, 1, sess->sess_ka * 1000)) == -1 ||
176: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
177: ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
178: return -1;
179: } else if (!ret) {
180: ioDEBUG(5, "Warning:: Session is abandoned ... must be disconnect!");
181: return 1;
182: }
183: /* receive & decode packet */
184: if (recv(sess->sess_sock, sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0) == -1) {
185: ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
186: return -1;
187: }
188: if (mqtt_readPINGRESP(sess->sess_buf)) {
189: ioDEBUG(5, "Warning:: Session is broken, not hear ping response ... must be disconnect!");
190: return 2;
191: }
192:
193: /* Keep Alive is OK! */
194: return 0;
195: }
196:
197: static void *
198: thrSession(struct tagSession *sess)
199: {
200: int ret, locKill = 42;
201: struct pollfd pfd;
202: struct mqtthdr *hdr;
203: ait_val_t *v;
204:
205: pthread_cleanup_push((void(*)(void*)) stopSession, sess);
206: ioTRACE(2);
207:
208: pfd.fd = sess->sess_sock;
209: pfd.events = POLLIN | POLLPRI;
210: while (!Kill && locKill) {
211: if ((ret = poll(&pfd, 1, sess->sess_ka * 1000)) == -1 ||
212: pfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
213: ioDEBUG(3, "Error:: poll(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
214: break;
215: } else if (!ret && (ret = KASession(sess))) {
216: call.LOG(logg, "Session %s keep-alive missing from %s for user %s ...\n",
217: sess->sess_cid, sess->sess_addr, sess->sess_user);
218: break;
219: }
220: /* receive & decode packet */
221: if ((ret = recv(sess->sess_sock, sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) {
222: ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
223: break;
224: } else if (!ret) {
225: ioDEBUG(4, "Session %s EOF received.", sess->sess_cid);
226: break;
227: } else
228: hdr = (struct mqtthdr*) sess->sess_buf->msg_base;
229:
230: /* dispatch message type */
1.2.2.3 misho 231: if (mqtt_srv_Dispatch(sess->sess_srv, sess))
232: ioLIBERR(mqtt);
1.2 misho 233: switch (hdr->mqtt_msg.type) {
234: case MQTT_TYPE_CONNECT:
235: ioDEBUG(5, "Exec CONNECT session");
236: if ((v = io_allocVar())) {
237: AIT_SET_STR(v, sess->sess_addr);
238: if (!schedEvent(root, startSession, v, (u_long) sess->sess_sock, sess, ret))
239: io_freeVar(v);
240: } else
241: ioLIBERR(mqtt);
1.2.2.13! misho 242:
! 243: SESS_LOCK;
! 244: TAILQ_REMOVE(&Sessions, sess, sess_node);
! 245: SESS_UNLOCK;
! 246:
! 247: locKill ^= locKill;
! 248: break;
1.2 misho 249: case MQTT_TYPE_DISCONNECT:
250: ioDEBUG(5, "Exec DISCONNECT session");
1.2.2.9 misho 251: finiSession(sess);
1.2.2.13! misho 252:
! 253: SESS_LOCK;
! 254: TAILQ_REMOVE(&Sessions, sess, sess_node);
! 255: SESS_UNLOCK;
! 256:
! 257: locKill ^= locKill;
1.2 misho 258: continue;
259: case MQTT_TYPE_PUBLISH:
260: ioDEBUG(5, "Exec PUBLISH topic QoS=%d", hdr->mqtt_msg.qos);
1.2.2.3 misho 261: /*
262: if (cmdPUBLISH(sess))
1.2 misho 263: locKill ^= locKill;
1.2.2.3 misho 264: */
1.2 misho 265: break;
266: case MQTT_TYPE_PUBREL:
267: break;
268: case MQTT_TYPE_SUBSCRIBE:
269: break;
270: case MQTT_TYPE_UNSUBSCRIBE:
271: break;
272: case MQTT_TYPE_PINGREQ:
1.2.2.4 misho 273: ioDEBUG(5, "Exec PINGREQ session");
1.2 misho 274: break;
1.2.2.11 misho 275: case MQTT_TYPE_PINGRESP:
276: ioDEBUG(5, "Exec PINGRESP session");
277: break;
1.2 misho 278: default:
279: ioDEBUG(5, "Error:: Session %s, wrong command %d - DISCARDED",
280: sess->sess_cid, hdr->mqtt_msg.type);
281: break;
282: }
283: }
284:
285: pthread_cleanup_pop(locKill);
286: pthread_exit(NULL);
1.2.2.11 misho 287: return NULL;
1.2 misho 288: }
289:
290: static void *
291: startSession(sched_task_t *task)
292: {
293: u_char basebuf[USHRT_MAX];
294: mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf };
295: mqtthdr_connflgs_t flg;
296: mqtthdr_connack_t cack;
297: struct tagSession *s, *sess = NULL;
298: int ret;
299:
300: ioTRACE(4);
301:
302: assert(task);
303:
304: if (!TASK_DATA(task)) {
1.2.2.8 misho 305: /* flow from accept new clients */
1.2 misho 306: sess = initSession(TASK_FD(task), TASK_ARG(task));
307: if (!sess) {
308: io_freeVar(TASK_ARG(task));
309: close(TASK_FD(task));
310: return NULL;
311: }
312:
313: /* receive & decode packet */
314: if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {
315: ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
1.2.2.9 misho 316: finiSession(sess);
1.2 misho 317: return NULL;
318: }
319: } else {
320: sess = TASK_DATA(task);
321: buf.msg_len = TASK_DATLEN(task) > sizeof basebuf ? sizeof basebuf : TASK_DATLEN(task);
322: memcpy(buf.msg_base, sess->sess_buf->msg_base, buf.msg_len);
323: }
324:
325: cack = mqtt_readCONNECT(&buf, &sess->sess_ka, sess->sess_cid, sizeof sess->sess_cid,
326: sess->sess_user, sizeof sess->sess_user, sess->sess_pass, sizeof sess->sess_pass,
327: &sess->sess_will.topic, &sess->sess_will.msg);
328: ret = cack.retcode;
329: flg.flags = cack.reserved;
330: if (flg.reserved) {
331: ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());
332: goto end;
333: } else {
334: sess->sess_clean = flg.clean_sess;
335: sess->sess_will.qos = flg.will_qos;
336: sess->sess_will.retain = flg.will_retain;
337: sess->sess_will.flag = flg.will_flg;
338: }
339:
340: /* check online table for user */
341: if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) {
342: ioDEBUG(0, "Login:: DENIED for username %s and password %s",
343: sess->sess_user, sess->sess_pass);
344: ret = MQTT_RETCODE_DENIED;
345: goto end;
346: } else {
347: ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user);
348: ret = MQTT_RETCODE_ACCEPTED;
349: }
1.2.2.12 misho 350:
1.2 misho 351: if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {
352: ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid);
353: TAILQ_FOREACH(s, &Sessions, sess_node)
354: if (!strcmp(s->sess_cid, sess->sess_cid)) {
355: /* found stale session & disconnect it! */
356: stopSession(s);
357: break;
358: }
359: }
1.2.2.12 misho 360:
1.2 misho 361: if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr,
362: sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg,
363: sess->sess_will.qos, sess->sess_will.retain) == -1) {
364: ioDEBUG(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user);
365: ret = MQTT_RETCODE_DENIED;
366: goto end;
367: } else {
368: ioDEBUG(0, "Session %s from %s and username %s is started",
369: sess->sess_cid, sess->sess_addr, sess->sess_user);
370: ret = MQTT_RETCODE_ACCEPTED;
371: }
372:
373: ret = mqtt_msgCONNACK(&msg, ret);
374: if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {
375: ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
1.2.2.9 misho 376: finiSession(sess);
1.2 misho 377: return NULL;
378: } else {
379: ioDEBUG(5, "Sended %d bytes", ret);
380: free(msg.msg_base);
381: memset(&msg, 0, sizeof msg);
382: }
383:
384: /* Start session thread OK ... */
1.2.2.6 misho 385: SESS_LOCK;
1.2 misho 386: TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);
1.2.2.12 misho 387: pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess);
1.2.2.6 misho 388: SESS_UNLOCK;
1.2 misho 389:
390: call.LOG(logg, "Session %s started from %s for user %s (timeout=%d) OK!\n", sess->sess_cid,
391: sess->sess_addr, sess->sess_user, sess->sess_ka);
392: return NULL;
393: end: /* close client connection */
394: ret = mqtt_msgCONNACK(&msg, ret);
395: if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {
396: ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
397: } else {
398: ioDEBUG(5, "Sended %d bytes", ret);
399: free(msg.msg_base);
400: memset(&msg, 0, sizeof msg);
401: }
402:
403: ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock);
1.2.2.9 misho 404: finiSession(sess);
1.2 misho 405: return NULL;
406: }
407:
408: static void *
1.2.2.9 misho 409: acceptClient(sched_task_t *task)
1.2 misho 410: {
1.2.2.9 misho 411: int cli;
412: io_sockaddr_t sa;
413: socklen_t sslen = sizeof sa.ss;
414: ait_val_t *v;
415: char str[STRSIZ];
1.2 misho 416:
1.2.2.9 misho 417: ioTRACE(4);
1.2 misho 418:
1.2.2.9 misho 419: assert(task);
1.2 misho 420:
1.2.2.9 misho 421: if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1)
422: goto end;
423: else
424: fcntl(TASK_FD(task), F_SETFL, fcntl(TASK_FD(task), F_GETFL, 0) | O_NONBLOCK);
425:
426: v = io_allocVar();
427: if (!v) {
428: ioLIBERR(io);
429: close(cli);
430: goto end;
431: } else {
432: memset(str, 0, sizeof str);
433: snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));
434: AIT_SET_STR(v, str);
435: }
436: ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));
437:
438: if (!schedRead(root, startSession, v, cli, NULL, 0)) {
439: io_freeVar(v);
440: close(cli);
441: ioDEBUG(1, "Terminated client with socket=%d", cli);
442: }
443: end:
1.2.2.11 misho 444: if (!schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0))
445: ioLIBERR(sched);
1.2.2.9 misho 446: return NULL;
1.2 misho 447: }
448:
1.2.2.9 misho 449: /* ----------------------------------------------------------------------- */
450:
1.2 misho 451: int
452: Run(int sock)
453: {
1.2.2.9 misho 454: struct tagPub *pub;
455: struct timespec pl = { 0, 100000000 };
1.2 misho 456:
457: ioTRACE(1);
458:
1.2.2.9 misho 459: if (listen(sock, SOMAXCONN) == -1) {
1.2 misho 460: ioSYSERR(0);
461: return -1;
462: } else
1.2.2.9 misho 463: fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) | O_NONBLOCK);
1.2 misho 464:
1.2.2.9 misho 465: /* state machine - accept new connections */
466: if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) {
467: ioLIBERR(sched);
1.2 misho 468: return -1;
469: }
470:
1.2.2.10 misho 471: pthread_attr_init(&attr);
472: pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
473:
1.2.2.9 misho 474: schedPolling(root, &pl, NULL);
475: schedRun(root, &Kill);
1.2 misho 476:
1.2.2.10 misho 477: pthread_attr_destroy(&attr);
478:
1.2.2.9 misho 479: /* free all undeleted elements into lists */
480: PUBS_LOCK;
481: TAILQ_FOREACH(pub, &Pubs, pub_node) {
482: TAILQ_REMOVE(&Pubs, pub, pub_node);
483:
484: AIT_FREE_VAL(&pub->pub_name);
485: if (pub->pub_packet.msg_base)
486: free(pub->pub_packet.msg_base);
1.2 misho 487: }
1.2.2.9 misho 488: PUBS_UNLOCK;
1.2 misho 489: return 0;
490: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to daemon.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