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