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

1.1.2.1   misho       1: #include "global.h"
1.1.2.16  misho       2: #include "rtlm.h"
1.1.2.11  misho       3: #include "mqttd.h"
1.1.2.1   misho       4: 
                      5: 
1.1.2.14  misho       6: static inline struct tagSession *
1.1.2.18  misho       7: initSession(int sock, ait_val_t * __restrict v)
1.1.2.14  misho       8: {
                      9:        struct tagSession *sess = NULL;
                     10:        const char *str;
                     11: 
1.1.2.20  misho      12:        ioTRACE(5);
1.1.2.15  misho      13: 
1.1.2.18  misho      14:        if (!v)
1.1.2.14  misho      15:                return NULL;
                     16: 
                     17:        sess = malloc(sizeof(struct tagSession));
                     18:        if (!sess) {
1.1.2.20  misho      19:                ioDEBUG(3, "Error:: in malloc #%d - %s", errno, strerror(errno));
1.1.2.18  misho      20:                io_freeVar(v);
1.1.2.14  misho      21:                return NULL;
                     22:        } else
                     23:                memset(sess, 0, sizeof(struct tagSession));
                     24: 
1.1.2.17  misho      25:        str = (const char*) cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry"));
1.1.2.14  misho      26:        if (!str)
                     27:                sess->sess_retry = DEFAULT_RETRY;
                     28:        else
                     29:                sess->sess_retry = strtol(str, NULL, 0);
                     30: 
                     31:        sess->sess_buf = mqtt_msgAlloc(0);
                     32:        if (!sess->sess_buf) {
1.1.2.20  misho      33:                ioDEBUG(3, "Error:: in msgAlloc #%d - %s", mqtt_GetErrno(), mqtt_GetError());
1.1.2.14  misho      34:                free(sess);
1.1.2.18  misho      35:                io_freeVar(v);
1.1.2.14  misho      36:                return NULL;
                     37:        }
                     38: 
1.1.2.15  misho      39:        sess->sess_sock = sock;
1.1.2.19  misho      40:        strlcpy(sess->sess_addr, (char*) AIT_GET_STR(v), sizeof sess->sess_addr);
1.1.2.18  misho      41:        io_freeVar(v);
1.1.2.14  misho      42:        return sess;
                     43: }
                     44: 
                     45: static void
                     46: finiSession(struct tagSession *sess)
                     47: {
                     48:        register int i;
                     49:        struct tagStore *store;
                     50: 
1.1.2.20  misho      51:        ioTRACE(5);
1.1.2.15  misho      52: 
1.1.2.14  misho      53:        if (!sess)
                     54:                return;
                     55: 
                     56:        for (i = 0; i < MQTT_QOS_RESERVED; i++)
                     57:                while ((store = SLIST_FIRST(&sess->sess_store[i]))) {
                     58:                        SLIST_REMOVE_HEAD(&sess->sess_store[i], st_node);
                     59:                        free(store);
                     60:                }
                     61: 
                     62:        if (sess->sess_will.msg)
                     63:                free(sess->sess_will.msg);
                     64:        if (sess->sess_will.topic)
                     65:                free(sess->sess_will.topic);
                     66: 
                     67:        if (sess->sess_sock > STDERR_FILENO) {
                     68:                shutdown(sess->sess_sock, SHUT_RDWR);
                     69:                close(sess->sess_sock);
                     70:        }
                     71: 
                     72:        mqtt_msgFree(&sess->sess_buf, 42);
                     73: 
                     74:        free(sess);
                     75: }
                     76: 
1.1.2.5   misho      77: static void *
1.1.2.15  misho      78: thrSession(struct tagSession *sess)
                     79: {
1.1.2.18  misho      80:        void thrClean(struct tagSession *sess)
                     81:        {
                     82:                pthread_mutex_lock(&mtx_sess);
                     83:                TAILQ_REMOVE(&Sessions, sess, sess_node);
                     84:                pthread_mutex_unlock(&mtx_sess);
1.1.2.20  misho      85:                ioDEBUG(1, "Close socket=%d", sess->sess_sock);
1.1.2.18  misho      86:                finiSession(sess);
                     87:        }
                     88: 
                     89:        pthread_cleanup_push((void(*)(void*)) thrClean, sess);
                     90: 
1.1.2.20  misho      91:        ioTRACE(2);
1.1.2.15  misho      92: 
1.1.2.18  misho      93:        pthread_cleanup_pop(42);
1.1.2.15  misho      94:        pthread_exit(NULL);
                     95: }
                     96: 
                     97: static void *
1.1.2.5   misho      98: startSession(sched_task_t *task)
1.1.2.4   misho      99: {
1.1.2.11  misho     100:        u_char basebuf[USHRT_MAX];
1.1.2.18  misho     101:        mqtt_msg_t msg = { NULL, 0 }, buf = { basebuf, sizeof basebuf };
1.1.2.10  misho     102:        mqtthdr_connflgs_t flg;
1.1.2.13  misho     103:        mqtthdr_connack_t cack;
1.1.2.12  misho     104:        struct tagSession *sess = NULL;
1.1.2.18  misho     105:        int ret;
                    106:        pthread_attr_t attr;
1.1.2.6   misho     107: 
1.1.2.20  misho     108:        ioTRACE(4);
1.1.2.6   misho     109: 
1.1.2.14  misho     110:        sess = initSession(TASK_FD(task), TASK_ARG(task));
1.1.2.15  misho     111:        if (!sess) {
1.1.2.18  misho     112:                io_freeVar(TASK_ARG(task));
1.1.2.15  misho     113:                close(TASK_FD(task));
1.1.2.14  misho     114:                return NULL;
1.1.2.15  misho     115:        }
1.1.2.6   misho     116: 
1.1.2.18  misho     117:        /* receive & decode packet */
1.1.2.14  misho     118:        if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {
1.1.2.20  misho     119:                ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
1.1.2.18  misho     120:                finiSession(sess);
                    121:                return NULL;
1.1.2.9   misho     122:        }
1.1.2.14  misho     123:        cack = mqtt_readCONNECT(&buf, &sess->sess_ka, sess->sess_cid, sizeof sess->sess_cid, 
1.1.2.12  misho     124:                        sess->sess_user, sizeof sess->sess_user, sess->sess_pass, sizeof sess->sess_pass, 
1.1.2.13  misho     125:                        &sess->sess_will.topic, &sess->sess_will.msg);
1.1.2.18  misho     126:        ret = cack.retcode;
1.1.2.14  misho     127:        flg.flags = cack.reserved;
                    128:        if (flg.reserved) {
1.1.2.20  misho     129:                ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());
1.1.2.10  misho     130:                goto end;
1.1.2.14  misho     131:        } else {
                    132:                sess->sess_clean = flg.clean_sess;
                    133:                sess->sess_will.qos = flg.will_qos;
                    134:                sess->sess_will.retain = flg.will_retain;
                    135:                sess->sess_will.flag = flg.will_flg;
1.1.2.12  misho     136:        }
1.1.2.6   misho     137: 
1.1.2.8   misho     138:        /* check online table for user */
1.1.2.18  misho     139:        if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) {
1.1.2.20  misho     140:                ioDEBUG(0, "Login:: DENIED for username %s and password %s", 
1.1.2.16  misho     141:                                sess->sess_user, sess->sess_pass);
1.1.2.21  misho     142:                ret = MQTT_RETCODE_DENIED;
1.1.2.16  misho     143:                goto end;
1.1.2.21  misho     144:        } else {
1.1.2.20  misho     145:                ioDEBUG(0, "Login:: ALLOWED for username %s ...", sess->sess_user);
1.1.2.21  misho     146:                ret = MQTT_RETCODE_ACCEPTED;
                    147:        }
1.1.2.23! misho     148:        if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {
        !           149:                // TODO: disconnect client
        !           150:        }
        !           151:        if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr, 
        !           152:                                sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg, 
        !           153:                                sess->sess_will.qos, sess->sess_will.retain) == -1) {
        !           154:                ioDEBUG(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user);
        !           155:                ret = MQTT_RETCODE_DENIED;
        !           156:                goto end;
        !           157:        } else {
        !           158:                ioDEBUG(0, "Session %s from %s and username %s is started", 
        !           159:                                sess->sess_cid, sess->sess_addr, sess->sess_user);
        !           160:                ret = MQTT_RETCODE_ACCEPTED;
        !           161:        }
1.1.2.6   misho     162: 
1.1.2.18  misho     163:        ret = mqtt_msgCONNACK(&msg, ret);
                    164:        if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {
1.1.2.20  misho     165:                ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
1.1.2.18  misho     166:                finiSession(sess);
                    167:                return NULL;
                    168:        } else {
1.1.2.20  misho     169:                ioDEBUG(5, "Sended %d bytes", ret);
1.1.2.19  misho     170:                free(msg.msg_base);
                    171:                msg.msg_len = 0;
1.1.2.18  misho     172:        }
                    173: 
1.1.2.15  misho     174:        /* Start session thread OK ... */
1.1.2.18  misho     175:        pthread_attr_init(&attr);
                    176:        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
                    177: 
                    178:        pthread_mutex_lock(&mtx_sess);
1.1.2.15  misho     179:        TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);
1.1.2.18  misho     180:        pthread_create(&sess->sess_tid, &attr, (void*(*)(void*)) thrSession, sess);
                    181:        pthread_mutex_unlock(&mtx_sess);
                    182: 
1.1.2.23! misho     183:        call.LOG(logg, "Session %s started from %s for user %s OK!\n", sess->sess_cid, 
        !           184:                        sess->sess_addr, sess->sess_user);
1.1.2.18  misho     185: 
                    186:        pthread_attr_destroy(&attr);
1.1.2.15  misho     187:        return NULL;
1.1.2.6   misho     188: end:   /* close client connection */
1.1.2.18  misho     189:        ret = mqtt_msgCONNACK(&msg, ret);
                    190:        if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {
1.1.2.20  misho     191:                ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
1.1.2.18  misho     192:        } else {
1.1.2.20  misho     193:                ioDEBUG(5, "Sended %d bytes", ret);
1.1.2.19  misho     194:                free(msg.msg_base);
                    195:                msg.msg_len = 0;
1.1.2.18  misho     196:        }
                    197: 
1.1.2.20  misho     198:        ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock);
1.1.2.15  misho     199:        finiSession(sess);
1.1.2.6   misho     200:        return NULL;
1.1.2.4   misho     201: }
                    202: 
                    203: /* ----------------------------------------------------------------------- */
                    204: 
1.1.2.5   misho     205: static void *
                    206: thrSched(void *arg __unused)
                    207: {
1.1.2.18  misho     208:        struct tagSession *sess;
1.1.2.5   misho     209: 
1.1.2.20  misho     210:        ioTRACE(1);
                    211: 
                    212:        schedRun(root, &Kill);
1.1.2.18  misho     213: 
                    214:        TAILQ_FOREACH(sess, &Sessions, sess_node)
                    215:                if (sess->sess_tid)
                    216:                        pthread_cancel(sess->sess_tid);
1.1.2.5   misho     217:        pthread_exit(NULL);
                    218: }
                    219: 
1.1.2.2   misho     220: int
1.1.2.3   misho     221: Run(int sock)
1.1.2.2   misho     222: {
1.1.2.4   misho     223:        io_sockaddr_t sa;
                    224:        socklen_t sslen = sizeof sa.ss;
                    225:        int cli;
1.1.2.5   misho     226:        pthread_t tid;
1.1.2.18  misho     227:        ait_val_t *v;
1.1.2.23! misho     228:        char str[STRSIZ];
1.1.2.4   misho     229: 
1.1.2.20  misho     230:        ioTRACE(1);
1.1.2.2   misho     231: 
1.1.2.5   misho     232:        if (pthread_create(&tid, NULL, thrSched, NULL)) {
1.1.2.20  misho     233:                ioLOGERR(0);
1.1.2.5   misho     234:                return -1;
                    235:        } else
                    236:                pthread_detach(tid);
1.1.2.20  misho     237:        ioDEBUG(2, "Run scheduler management thread");
1.1.2.5   misho     238: 
1.1.2.20  misho     239:        if (listen(sock, 5 /*SOMAXCONN*/) == -1) {
                    240:                ioLOGERR(0);
1.1.2.3   misho     241:                return -1;
                    242:        }
                    243: 
1.1.2.4   misho     244:        while (!Kill) {
                    245:                if ((cli = accept(sock, &sa.sa, &sslen)) == -1) {
1.1.2.20  misho     246:                        if (!Kill)
                    247:                                continue;
                    248:                        ioLOGERR(0);
                    249:                        break;
1.1.2.18  misho     250:                }
                    251:                v = io_allocVar();
                    252:                if (!v) {
                    253:                        ioLIBERR(mqtt);
                    254:                        break;
1.1.2.23! misho     255:                } else {
        !           256:                        memset(str, 0, sizeof str);
        !           257:                        snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));
        !           258:                        AIT_SET_STR(v, str);
        !           259:                }
        !           260:                ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));
1.1.2.4   misho     261: 
1.1.2.18  misho     262:                if (!schedRead(root, startSession, v, cli)) {
                    263:                        io_freeVar(v);
1.1.2.4   misho     264:                        close(cli);
1.1.2.20  misho     265:                        ioDEBUG(1, "Terminated client with socket=%d", cli);
1.1.2.4   misho     266:                }
                    267:        }
                    268: 
1.1.2.2   misho     269:        return 0;
                    270: }

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