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