#include "global.h"
#include "mqttd.h"
extern char cliCmd[], *cliStr[];
static void *
startSession(sched_task_t *task)
{
u_char basebuf[USHRT_MAX];
mqtt_cb_t cbs[MQTT_TYPE_MAX + 1] = { 0 };
// mqtt_msg_t *buf = { basebuf, sizeof basebuf };
mqtthdr_connflgs_t flg;
int ret = 0;
struct timeval tv = { 0 };
u_short ka;
FTRACE(4);
/*
buf = mqtt_msgAlloc(USHRT_MAX);
if (!buf) {
syslog(LOG_ERR, "Error:: allocate message buf (%d) #%d - %s", (int) TASK_FD(task),
mqtt_GetErrno(), mqtt_GetError());
goto end;
}
*/
if (recv(TASK_FD(task), basebuf, sizeof basebuf, 0) == -1) {
VERB(3) syslog(LOG_ERR, "Error:: recv(%d) #%d - %s", (int) TASK_FD(task),
errno, strerror(errno));
goto end;
}
/*
flg = mqtt_readCONNECT(buf, &ka, );
if (flg.reserved) {
VERB(3) syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s",
mqtt_GetErrno(), mqtt_GetError());
goto end;
} */
/*
for (ret = i = 0; cliCmd[i] && !(ret = (hdr->mqtt_msg.type == cliCmd[i])); i++);
if (!ret) {
VERB(2) syslog(LOG_ERR, "Error:: wrong command type #%d %s",
hdr->mqtt_msg.type, cliStr[i]);
goto end;
}
*/
/* check online table for user */
// ChkSessPUB(&cfg, );
// ret = mqttDispatcher(cbs, buf);
// mqtt_msgFree(&buf, 42);
/*
switch (ret) {
case -1:
syslog(LOG_ERR, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());
break;
default:
return NULL;
}
*/
end: /* close client connection */
close(TASK_FD(task));
VERB(1) syslog(LOG_DEBUG, "Close client %s with socket=%d",
(char*) TASK_ARG(task), (int) TASK_FD(task));
if (TASK_ARG(task))
free(TASK_ARG(task));
return NULL;
}
/* ----------------------------------------------------------------------- */
static void *
thrSched(void *arg __unused)
{
FTRACE(1);
schedRun(root, (intptr_t*) &Kill);
pthread_exit(NULL);
}
int
Run(int sock)
{
io_sockaddr_t sa;
socklen_t sslen = sizeof sa.ss;
int cli;
char *str = NULL, szAddr[STRSIZ] = { 0 };
pthread_t tid;
FTRACE(1);
if (pthread_create(&tid, NULL, thrSched, NULL)) {
syslog(LOG_ERR, "Error:: thread scheduler #%d - %s", errno, strerror(errno));
return -1;
} else
pthread_detach(tid);
VERB(2) syslog(LOG_DEBUG, "Run scheduler management thread");
if (listen(sock, SOMAXCONN) == -1) {
syslog(LOG_ERR, "Error:: listen(%d) #%d - %s\n", sock, errno, strerror(errno));
pthread_cancel(tid);
return -1;
}
while (!Kill) {
if ((cli = accept(sock, &sa.sa, &sslen)) == -1) {
syslog(LOG_ERR, "Error:: accept() #%d - %s", errno, strerror(errno));
continue;
} else
VERB(1) {
switch (sa.sa.sa_family) {
case AF_INET:
inet_ntop(AF_INET, &sa.sin.sin_addr, szAddr, sslen);
snprintf(szAddr, sizeof szAddr, "%s:%d",
szAddr, ntohs(sa.sin.sin_port));
break;
case AF_INET6:
inet_ntop(AF_INET6, &sa.sin6.sin6_addr, szAddr, sslen);
snprintf(szAddr, sizeof szAddr, "%s:%d",
szAddr, ntohs(sa.sin6.sin6_port));
break;
case AF_LOCAL:
strlcpy(szAddr, sa.sun.sun_path, sizeof szAddr);
break;
default:
close(cli);
syslog(LOG_ERR, "Error:: unsupported address type %d",
sa.sa.sa_family);
continue;
}
str = strdup(szAddr);
syslog(LOG_DEBUG, "Connected client with socket=%d from %s", cli, str);
}
if (!schedRead(root, startSession, str, cli)) {
close(cli);
VERB(1) syslog(LOG_DEBUG, "Terminated client with socket=%d", cli);
if (str)
free(str);
}
}
return 0;
}
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>