Annotation of mqtt/src/daemon.c, revision 1.2.2.19

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

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>