--- mqtt/src/mqttd_calls.c 2012/07/03 12:46:01 1.4 +++ mqtt/src/mqttd_calls.c 2017/10/08 22:49:25 1.5 @@ -3,7 +3,7 @@ * by Michael Pounov * * $Author: misho $ -* $Id: mqttd_calls.c,v 1.4 2012/07/03 12:46:01 misho Exp $ +* $Id: mqttd_calls.c,v 1.5 2017/10/08 22:49:25 misho Exp $ * ************************************************************************** The ELWIX and AITNET software is distributed under the following @@ -12,7 +12,7 @@ terms: All of the documentation and software included in the ELWIX and AITNET Releases is copyrighted by ELWIX - Sofia/Bulgaria -Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 by Michael Pounov . All rights reserved. Redistribution and use in source and binary forms, with or without @@ -55,8 +55,9 @@ mkPkt(void * __restrict data, int dlen) { ait_val_t *p = NULL; - if (!(p = io_allocVar())) { - ioDEBUG(7, "Error:: in send packet prepare #%d - %s", io_GetErrno(), io_GetError()); + if (!(p = ait_allocVar())) { + EVERBOSE(7, "Error:: in send packet prepare #%d - %s", + elwix_GetErrno(), elwix_GetError()); return NULL; } @@ -72,7 +73,7 @@ freePkt(ait_val_t ** __restrict p) if (!p) return; - io_freeVar(p); + ait_freeVar(p); } static void * @@ -83,16 +84,16 @@ sendPacket(sched_task_t *task) u_char *pos; if (!p || AIT_ISEMPTY(p)) { - ioDEBUG(9, "Error:: invalid packet or found empty content ..."); + EVERBOSE(9, "Error:: invalid packet or found empty content ..."); return NULL; } - ioDEBUG(7, "Send packet length %d for socket %d\n", AIT_LEN(p), (u_int) TASK_FD(task)); + EVERBOSE(7, "Send packet length %d for socket %d\n", AIT_LEN(p), (u_int) TASK_FD(task)); for (slen = AIT_LEN(p), pos = AIT_GET_BUF(p); slen > 0; slen -= n, pos += n) { n = send(TASK_FD(task), pos, slen, MSG_NOSIGNAL); if (n == -1) { - ioSYSERR(0); + ESYSERR(0); break; } } @@ -121,7 +122,7 @@ search4send(struct tagSession * __restrict sess, const if ((ret = regcomp(&re, st->st_subscr.sub_topic.msg_base, REG_EXTENDED))) { regerror(ret, &re, szStr, sizeof szStr); regfree(&re); - ioDEBUG(3, "Error:: regcomp(%s) %s\n", (char*) + EVERBOSE(3, "Error:: regcomp(%s) %s\n", (char*) st->st_subscr.sub_topic.msg_base, szStr); } if (!regexec(&re, topic, 1, &match, 0)) { @@ -147,7 +148,7 @@ sendRetain(sched_task_t *task) struct tagSession *sess; int siz; - ioTRACE(2); + ETRACE(); assert(task); @@ -155,7 +156,7 @@ sendRetain(sched_task_t *task) assert(sess); if (!sess->sess_buf) { - ioDEBUG(9, "WARNING! No allocated buffer!?!\n"); + EVERBOSE(9, "WARNING! No allocated buffer!?!\n"); return NULL; } @@ -167,7 +168,7 @@ sendRetain(sched_task_t *task) siz = s->sub_value.msg_len; memcpy(sess->sess_buf->msg_base, s->sub_value.msg_base, MIN(sess->sess_buf->msg_len, s->sub_value.msg_len)); - ioDEBUG(7, "Sending retain message %d bytes, QoS %hhd topic '%s' data length %d\n", + EVERBOSE(7, "Sending retain message %d bytes, QoS %hhd topic '%s' data length %d\n", siz, s->sub_ret, (char*) s->sub_topic.msg_base, s->sub_value.msg_len); if (siz > 0) search4send(sess, s->sub_topic.msg_base, siz, s->sub_ret); @@ -182,7 +183,7 @@ pubWill(struct tagSession * __restrict sess) { int datlen; - ioTRACE(2); + ETRACE(); /* prepare will packet */ datlen = mqtt_msgPUBLISH(sess->sess_buf, sess->sess_will.topic, 0xDEAD, 0, 1, 0, @@ -245,15 +246,15 @@ cmdPUBLISH(void *srv, int len, void *arg) u_short mid = 0; ait_val_t *p = NULL; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec PUBLISH session"); + EVERBOSE(5, "Exec PUBLISH session"); siz = mqtt_readPUBLISH(sess->sess_buf, szTopic, sizeof szTopic, &mid, NULL); if (siz == -1) { - ioDEBUG(5, "Error:: in readPUBLISH #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in readPUBLISH #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -264,7 +265,7 @@ cmdPUBLISH(void *srv, int len, void *arg) return 0; siz = mqtt_msgPUBACK(sess->sess_buf, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgPUBACK #%d - %s", + EVERBOSE(5, "Error:: in msgPUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -274,7 +275,7 @@ cmdPUBLISH(void *srv, int len, void *arg) return 0; siz = mqtt_msgPUBREC(sess->sess_buf, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgPUBREC #%d - %s", + EVERBOSE(5, "Error:: in msgPUBREC #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -299,15 +300,15 @@ cmdPUBREL(void *srv, int len, void *arg) u_short mid = 0; ait_val_t *p = NULL; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec PUBREL session"); + EVERBOSE(5, "Exec PUBREL session"); mid = mqtt_readPUBREL(sess->sess_buf); if (mid == (u_short) -1) { - ioDEBUG(5, "Error:: in readPUBREL #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in readPUBREL #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -317,7 +318,7 @@ cmdPUBREL(void *srv, int len, void *arg) siz = mqtt_msgPUBCOMP(sess->sess_buf, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgPUBCOMP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in msgPUBCOMP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -340,23 +341,23 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) void *ptr; ait_val_t *p = NULL; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec SUBSCRIBE session"); + EVERBOSE(5, "Exec SUBSCRIBE session"); siz = mqtt_readSUBSCRIBE(sess->sess_buf, &mid, &subs); if (siz == -1) { - ioDEBUG(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in readSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } /* add to db */ for (i = 0; i < siz; i++) { - store = io_malloc(sizeof(struct tagStore)); + store = e_malloc(sizeof(struct tagStore)); if (!store) { - ioSYSERR(0); + ELIBERR(elwix); continue; } else { store->st_msgid = mid; @@ -369,11 +370,11 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) /* convert topic to regexp */ if (mqtt_expandTopic(subs[i].sub_topic.msg_base, buf, sizeof buf, 1, 1) == -1) { - ioDEBUG(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in regexp #%d - %s", mqtt_GetErrno(), mqtt_GetError()); } else { ptr = realloc(store->st_subscr.sub_topic.msg_base, strlen(buf) + 1); if (!ptr) { - ioSYSERR(0); + ESYSERR(0); continue; } else { store->st_subscr.sub_topic.msg_base = ptr; @@ -398,7 +399,7 @@ cmdSUBSCRIBE(void *srv, int len, void *arg) /* send acknowledge */ siz = mqtt_msgSUBACK(sess->sess_buf, subs, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in msgSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); goto end; } else { p = mkPkt(sess->sess_buf->msg_base, siz); @@ -422,15 +423,15 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) struct tagStore *store, *tmp; ait_val_t *p = NULL; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec UNSUBSCRIBE session"); + EVERBOSE(5, "Exec UNSUBSCRIBE session"); siz = mqtt_readUNSUBSCRIBE(sess->sess_buf, &mid, &subs); if (siz == -1) { - ioDEBUG(5, "Error:: in readUNSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in readUNSUBSCRIBE #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } @@ -447,7 +448,7 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) free(store->st_subscr.sub_topic.msg_base); if (store->st_subscr.sub_value.msg_base) free(store->st_subscr.sub_value.msg_base); - io_free(store); + e_free(store); } } @@ -458,7 +459,7 @@ cmdUNSUBSCRIBE(void *srv, int len, void *arg) /* send acknowledge */ siz = mqtt_msgUNSUBACK(sess->sess_buf, mid); if (siz == -1) { - ioDEBUG(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in msgUNSUBACK #%d - %s", mqtt_GetErrno(), mqtt_GetError()); goto end; } else { p = mkPkt(sess->sess_buf->msg_base, siz); @@ -478,15 +479,15 @@ cmdPINGREQ(void *srv, int len, void *arg) int siz = 0; ait_val_t *p = NULL; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec PINGREQ session"); + EVERBOSE(5, "Exec PINGREQ session"); siz = mqtt_msgPINGRESP(sess->sess_buf); if (siz == -1) { - ioDEBUG(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); + EVERBOSE(5, "Error:: in msgPINGRESP #%d - %s", mqtt_GetErrno(), mqtt_GetError()); return 0; } else { p = mkPkt(sess->sess_buf->msg_base, siz); @@ -503,12 +504,12 @@ cmdCONNECT(void *srv, int len, void *arg) struct tagStore *store; struct tagSession *sess = (struct tagSession*) arg; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec CONNECT session"); + EVERBOSE(5, "Exec CONNECT session"); TAILQ_REMOVE(&Sessions, sess, sess_node); schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL); @@ -530,7 +531,7 @@ cmdCONNECT(void *srv, int len, void *arg) if (store->st_subscr.sub_value.msg_base) free(store->st_subscr.sub_value.msg_base); - io_free(store); + e_free(store); } if (sess->sess_will.flag) @@ -552,12 +553,12 @@ cmdDISCONNECT(void *srv, int len, void *arg) { struct tagSession *sess = (struct tagSession*) arg; - ioTRACE(2); + ETRACE(); if (!sess) return -1; - ioDEBUG(5, "Exec DISCONNECT session"); + EVERBOSE(5, "Exec DISCONNECT session"); call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid, sess->sess_addr, sess->sess_user);