Annotation of mqtt/src/daemon.c, revision 1.1.2.14
1.1.2.1 misho 1: #include "global.h"
1.1.2.11 misho 2: #include "mqttd.h"
1.1.2.1 misho 3:
4:
1.1.2.7 misho 5: extern char cliCmd[], *cliStr[];
6:
7:
1.1.2.14! misho 8: static inline struct tagSession *
! 9: initSession(int sock, char * __restrict addr)
! 10: {
! 11: struct tagSession *sess = NULL;
! 12: const char *str;
! 13:
! 14: if (!addr)
! 15: return NULL;
! 16:
! 17: sess = malloc(sizeof(struct tagSession));
! 18: if (!sess) {
! 19: VERB(3) syslog(LOG_ERR, "Error:: in malloc #%d - %s", errno, strerror(errno));
! 20: free(addr);
! 21: return NULL;
! 22: } else
! 23: memset(sess, 0, sizeof(struct tagSession));
! 24:
! 25: str = cfg_GetAttribute(&cfg, CFG("mqttd"), CFG("retry"));
! 26: if (!str)
! 27: sess->sess_retry = DEFAULT_RETRY;
! 28: else
! 29: sess->sess_retry = strtol(str, NULL, 0);
! 30:
! 31: sess->sess_sock = sock;
! 32: strlcpy(sess->sess_addr, addr, sizeof sess->sess_addr);
! 33: free(addr);
! 34:
! 35: sess->sess_buf = mqtt_msgAlloc(0);
! 36: if (!sess->sess_buf) {
! 37: VERB(3) syslog(LOG_ERR, "Error:: in msgAlloc #%d - %s", mqtt_GetErrno(), mqtt_GetError());
! 38: free(sess);
! 39: return NULL;
! 40: }
! 41:
! 42: return sess;
! 43: }
! 44:
! 45: static void
! 46: finiSession(struct tagSession *sess)
! 47: {
! 48: register int i;
! 49: struct tagStore *store;
! 50:
! 51: if (!sess)
! 52: return;
! 53:
! 54: for (i = 0; i < MQTT_QOS_RESERVED; i++)
! 55: while ((store = SLIST_FIRST(&sess->sess_store[i]))) {
! 56: SLIST_REMOVE_HEAD(&sess->sess_store[i], st_node);
! 57: free(store);
! 58: }
! 59:
! 60: if (sess->sess_will.msg)
! 61: free(sess->sess_will.msg);
! 62: if (sess->sess_will.topic)
! 63: free(sess->sess_will.topic);
! 64:
! 65: if (sess->sess_sock > STDERR_FILENO) {
! 66: shutdown(sess->sess_sock, SHUT_RDWR);
! 67: close(sess->sess_sock);
! 68: }
! 69:
! 70: mqtt_msgFree(&sess->sess_buf, 42);
! 71:
! 72: if (sess->sess_tid)
! 73: pthread_cancel(sess->sess_tid);
! 74: free(sess);
! 75: }
! 76:
1.1.2.5 misho 77: static void *
78: startSession(sched_task_t *task)
1.1.2.4 misho 79: {
1.1.2.11 misho 80: u_char basebuf[USHRT_MAX];
1.1.2.6 misho 81: mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 };
1.1.2.12 misho 82: mqtt_msg_t buf = { basebuf, sizeof basebuf };
1.1.2.10 misho 83: mqtthdr_connflgs_t flg;
1.1.2.13 misho 84: mqtthdr_connack_t cack;
1.1.2.9 misho 85: int ret = 0;
1.1.2.10 misho 86: struct timeval tv = { 0 };
1.1.2.12 misho 87: struct tagSession *sess = NULL;
1.1.2.6 misho 88:
89: FTRACE(4);
90:
1.1.2.14! misho 91: sess = initSession(TASK_FD(task), TASK_ARG(task));
! 92: if (!sess)
! 93: return NULL;
1.1.2.6 misho 94:
1.1.2.14! misho 95: if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {
! 96: VERB(3) syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
1.1.2.6 misho 97: goto end;
1.1.2.9 misho 98: }
1.1.2.14! misho 99: cack = mqtt_readCONNECT(&buf, &sess->sess_ka, sess->sess_cid, sizeof sess->sess_cid,
1.1.2.12 misho 100: sess->sess_user, sizeof sess->sess_user, sess->sess_pass, sizeof sess->sess_pass,
1.1.2.13 misho 101: &sess->sess_will.topic, &sess->sess_will.msg);
1.1.2.14! misho 102: flg.flags = cack.reserved;
! 103: if (flg.reserved) {
1.1.2.13 misho 104: VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());
1.1.2.10 misho 105: goto end;
1.1.2.14! misho 106: } else {
! 107: sess->sess_clean = flg.clean_sess;
! 108: sess->sess_will.qos = flg.will_qos;
! 109: sess->sess_will.retain = flg.will_retain;
! 110: sess->sess_will.flag = flg.will_flg;
1.1.2.12 misho 111: }
1.1.2.6 misho 112:
1.1.2.8 misho 113: /* check online table for user */
1.1.2.6 misho 114:
115: end: /* close client connection */
1.1.2.12 misho 116: if (sess)
117: free(sess);
1.1.2.5 misho 118: close(TASK_FD(task));
119: VERB(1) syslog(LOG_DEBUG, "Close client %s with socket=%d",
120: (char*) TASK_ARG(task), (int) TASK_FD(task));
121: if (TASK_ARG(task))
122: free(TASK_ARG(task));
1.1.2.6 misho 123: return NULL;
1.1.2.4 misho 124: }
125:
126: /* ----------------------------------------------------------------------- */
127:
1.1.2.5 misho 128: static void *
129: thrSched(void *arg __unused)
130: {
131: FTRACE(1);
132:
133: schedRun(root, (intptr_t*) &Kill);
134: pthread_exit(NULL);
135: }
136:
1.1.2.2 misho 137: int
1.1.2.3 misho 138: Run(int sock)
1.1.2.2 misho 139: {
1.1.2.4 misho 140: io_sockaddr_t sa;
141: socklen_t sslen = sizeof sa.ss;
142: int cli;
1.1.2.5 misho 143: char *str = NULL, szAddr[STRSIZ] = { 0 };
144: pthread_t tid;
1.1.2.4 misho 145:
1.1.2.2 misho 146: FTRACE(1);
147:
1.1.2.5 misho 148: if (pthread_create(&tid, NULL, thrSched, NULL)) {
149: syslog(LOG_ERR, "Error:: thread scheduler #%d - %s", errno, strerror(errno));
150: return -1;
151: } else
152: pthread_detach(tid);
153: VERB(2) syslog(LOG_DEBUG, "Run scheduler management thread");
154:
1.1.2.3 misho 155: if (listen(sock, SOMAXCONN) == -1) {
1.1.2.4 misho 156: syslog(LOG_ERR, "Error:: listen(%d) #%d - %s\n", sock, errno, strerror(errno));
1.1.2.5 misho 157: pthread_cancel(tid);
1.1.2.3 misho 158: return -1;
159: }
160:
1.1.2.4 misho 161: while (!Kill) {
162: if ((cli = accept(sock, &sa.sa, &sslen)) == -1) {
163: syslog(LOG_ERR, "Error:: accept() #%d - %s", errno, strerror(errno));
164: continue;
165: } else
166: VERB(1) {
167: switch (sa.sa.sa_family) {
168: case AF_INET:
169: inet_ntop(AF_INET, &sa.sin.sin_addr, szAddr, sslen);
170: snprintf(szAddr, sizeof szAddr, "%s:%d",
171: szAddr, ntohs(sa.sin.sin_port));
172: break;
173: case AF_INET6:
174: inet_ntop(AF_INET6, &sa.sin6.sin6_addr, szAddr, sslen);
175: snprintf(szAddr, sizeof szAddr, "%s:%d",
176: szAddr, ntohs(sa.sin6.sin6_port));
177: break;
178: case AF_LOCAL:
179: strlcpy(szAddr, sa.sun.sun_path, sizeof szAddr);
180: break;
181: default:
182: close(cli);
183: syslog(LOG_ERR, "Error:: unsupported address type %d",
184: sa.sa.sa_family);
185: continue;
186: }
1.1.2.5 misho 187: str = strdup(szAddr);
188: syslog(LOG_DEBUG, "Connected client with socket=%d from %s", cli, str);
1.1.2.4 misho 189: }
190:
1.1.2.5 misho 191: if (!schedRead(root, startSession, str, cli)) {
1.1.2.4 misho 192: close(cli);
193: VERB(1) syslog(LOG_DEBUG, "Terminated client with socket=%d", cli);
1.1.2.5 misho 194: if (str)
195: free(str);
1.1.2.4 misho 196: }
197: }
198:
1.1.2.2 misho 199: return 0;
200: }
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