1: #include "global.h"
2:
3:
4: extern char cliCmd[], *cliStr[];
5:
6:
7: static void *
8: startSession(sched_task_t *task)
9: {
10: static u_char basebuf[USHRT_MAX];
11: mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 };
12: mqtt_msg_t *buf = { basebuf, sizeof basebuf };
13: mqtthdr_connflgs_t flg;
14: int ret = 0;
15: struct timeval tv = { 0 };
16: u_short ka;
17:
18: FTRACE(4);
19:
20: /*
21: buf = mqtt_msgAlloc(USHRT_MAX);
22: if (!buf) {
23: syslog(LOG_ERR, "Error:: allocate message buf (%d) #%d - %s", (int) TASK_FD(task),
24: mqtt_GetErrno(), mqtt_GetError());
25: goto end;
26: }
27: */
28:
29: if (recv(TASK_FD(task), basebuf, sizeof basebuf, 0) == -1) {
30: VERB(3) syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", (int) TASK_FD(task),
31: errno, strerror(errno));
32: goto end;
33: }
34:
35: flg = mqtt_readCONNECT(buf, &ka, );
36: if (flg.reserved) {
37: VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s",
38: mqtt_GetErrno(), mqtt_GetError());
39: goto end;
40: }
41:
42: /*
43: for (ret = i = 0; cliCmd[i] && !(ret = (hdr->mqtt_msg.type == cliCmd[i])); i++);
44: if (!ret) {
45: VERB(2) syslog(LOG_ERR, "Error:: wrong command type #%d %s",
46: hdr->mqtt_msg.type, cliStr[i]);
47: goto end;
48: }
49: */
50:
51: /* check online table for user */
52: // ChkSessPUB(&cfg, );
53:
54: // ret = mqttDispatcher(cbs, buf);
55: // mqtt_msgFree(&buf, 42);
56:
57: /*
58: switch (ret) {
59: case -1:
60: syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());
61: break;
62: default:
63: return NULL;
64: }
65: */
66: end: /* close client connection */
67: close(TASK_FD(task));
68: VERB(1) syslog(LOG_DEBUG, "Close client %s with socket=%d",
69: (char*) TASK_ARG(task), (int) TASK_FD(task));
70: if (TASK_ARG(task))
71: free(TASK_ARG(task));
72: return NULL;
73: }
74:
75: /* ----------------------------------------------------------------------- */
76:
77: static void *
78: thrSched(void *arg __unused)
79: {
80: FTRACE(1);
81:
82: schedRun(root, (intptr_t*) &Kill);
83: pthread_exit(NULL);
84: }
85:
86: int
87: Run(int sock)
88: {
89: io_sockaddr_t sa;
90: socklen_t sslen = sizeof sa.ss;
91: int cli;
92: char *str = NULL, szAddr[STRSIZ] = { 0 };
93: pthread_t tid;
94:
95: FTRACE(1);
96:
97: if (pthread_create(&tid, NULL, thrSched, NULL)) {
98: syslog(LOG_ERR, "Error:: thread scheduler #%d - %s", errno, strerror(errno));
99: return -1;
100: } else
101: pthread_detach(tid);
102: VERB(2) syslog(LOG_DEBUG, "Run scheduler management thread");
103:
104: if (listen(sock, SOMAXCONN) == -1) {
105: syslog(LOG_ERR, "Error:: listen(%d) #%d - %s\n", sock, errno, strerror(errno));
106: pthread_cancel(tid);
107: return -1;
108: }
109:
110: while (!Kill) {
111: if ((cli = accept(sock, &sa.sa, &sslen)) == -1) {
112: syslog(LOG_ERR, "Error:: accept() #%d - %s", errno, strerror(errno));
113: continue;
114: } else
115: VERB(1) {
116: switch (sa.sa.sa_family) {
117: case AF_INET:
118: inet_ntop(AF_INET, &sa.sin.sin_addr, szAddr, sslen);
119: snprintf(szAddr, sizeof szAddr, "%s:%d",
120: szAddr, ntohs(sa.sin.sin_port));
121: break;
122: case AF_INET6:
123: inet_ntop(AF_INET6, &sa.sin6.sin6_addr, szAddr, sslen);
124: snprintf(szAddr, sizeof szAddr, "%s:%d",
125: szAddr, ntohs(sa.sin6.sin6_port));
126: break;
127: case AF_LOCAL:
128: strlcpy(szAddr, sa.sun.sun_path, sizeof szAddr);
129: break;
130: default:
131: close(cli);
132: syslog(LOG_ERR, "Error:: unsupported address type %d",
133: sa.sa.sa_family);
134: continue;
135: }
136: str = strdup(szAddr);
137: syslog(LOG_DEBUG, "Connected client with socket=%d from %s", cli, str);
138: }
139:
140: if (!schedRead(root, startSession, str, cli)) {
141: close(cli);
142: VERB(1) syslog(LOG_DEBUG, "Terminated client with socket=%d", cli);
143: if (str)
144: free(str);
145: }
146: }
147:
148: return 0;
149: }
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>