mqtt/src/daemon.c - diff

Return to daemon.c CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / mqtt / src

Diff for /mqtt/src/daemon.c between versions 1.5 and 1.5.2.2

version 1.5, 2013/01/18 09:43:27	version 1.5.2.2, 2013/07/16 14:34:58
Line 12 terms:	Line 12 terms:
All of the documentation and software included in the ELWIX and AITNET	All of the documentation and software included in the ELWIX and AITNET
Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>	Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>

Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012	Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013
by Michael Pounov <misho@elwix.org>. All rights reserved.	by Michael Pounov <misho@elwix.org>. All rights reserved.

Redistribution and use in source and binary forms, with or without	Redistribution and use in source and binary forms, with or without
Line 56 initSession(int sock, ait_val_t * __restrict v)	Line 56 initSession(int sock, ait_val_t * __restrict v)
struct tagSession *sess = NULL;	struct tagSession *sess = NULL;
const char *str;	const char *str;

ioTRACE(5);	ETRACE();

if (!v)	if (!v)
return NULL;	return NULL;

sess = io_malloc(sizeof(struct tagSession));	sess = e_malloc(sizeof(struct tagSession));
if (!sess) {	if (!sess) {
ioSYSERR(0);	ELIBERR(elwix);
return NULL;	return NULL;
} else	} else
memset(sess, 0, sizeof(struct tagSession));	memset(sess, 0, sizeof(struct tagSession));
Line 78 initSession(int sock, ait_val_t * __restrict v)	Line 78 initSession(int sock, ait_val_t * __restrict v)

sess->sess_buf = mqtt_msgAlloc(USHRT_MAX);	sess->sess_buf = mqtt_msgAlloc(USHRT_MAX);
if (!sess->sess_buf) {	if (!sess->sess_buf) {
ioLIBERR(mqtt);	ELIBERR(mqtt);
io_free(sess);	e_free(sess);
return NULL;	return NULL;
}	}

/* init server actor */	/* init server actor */
sess->sess_srv = mqtt_srv_cliInit(sock, sess->sess_buf, sess->sess_ka, 1);	sess->sess_srv = mqtt_srv_cliInit(sock, sess->sess_buf, sess->sess_ka, 1);
if (!sess->sess_srv) {	if (!sess->sess_srv) {
ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());	EVERBOSE(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
mqtt_msgFree(&sess->sess_buf, 42);	mqtt_msgFree(&sess->sess_buf, 42);
io_free(sess);	e_free(sess);
return NULL;	return NULL;
} else {	} else {
mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT);	mqtt_srv_setCmd(sess->sess_srv, MQTT_TYPE_CONNECT, cmdCONNECT);
Line 110 finiSession(struct tagSession *sess)	Line 110 finiSession(struct tagSession *sess)
{	{
struct tagStore *store;	struct tagStore *store;

ioTRACE(5);	ETRACE();

if (!sess)	if (!sess)
return;	return;
Line 134 finiSession(struct tagSession *sess)	Line 134 finiSession(struct tagSession *sess)
if (store->st_subscr.sub_value.msg_base)	if (store->st_subscr.sub_value.msg_base)
free(store->st_subscr.sub_value.msg_base);	free(store->st_subscr.sub_value.msg_base);

io_free(store);	e_free(store);
}	}

if (sess->sess_will.flag)	if (sess->sess_will.flag)
Line 151 finiSession(struct tagSession *sess)	Line 151 finiSession(struct tagSession *sess)
mqtt_srv_cliFini(&sess->sess_srv);	mqtt_srv_cliFini(&sess->sess_srv);
mqtt_msgFree(&sess->sess_buf, 42);	mqtt_msgFree(&sess->sess_buf, 42);

io_free(sess);	e_free(sess);
}	}

static void *	static void *
Line 160 leaveClient(sched_task_t *task)	Line 160 leaveClient(sched_task_t *task)
struct tagSession *sess;	struct tagSession *sess;
int ret;	int ret;

ioTRACE(4);	ETRACE();

assert(task);	assert(task);

Line 172 leaveClient(sched_task_t *task)	Line 172 leaveClient(sched_task_t *task)
ret = mqtt_msgDISCONNECT(sess->sess_buf);	ret = mqtt_msgDISCONNECT(sess->sess_buf);
send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL);	send(TASK_FD(task), sess->sess_buf->msg_base, ret, MSG_NOSIGNAL);

ioDEBUG(1, "Close socket=%ld", (long) TASK_FD(task));	EVERBOSE(1, "Close socket=%ld", (long) TASK_FD(task));
call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,	call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,
sess->sess_addr, sess->sess_user);	sess->sess_addr, sess->sess_user);

Line 186 dispatchSession(sched_task_t *task)	Line 186 dispatchSession(sched_task_t *task)
int ret, len = 0;	int ret, len = 0;
struct tagSession *sess;	struct tagSession *sess;

ioTRACE(2);	ETRACE();

assert(task);	assert(task);

Line 195 dispatchSession(sched_task_t *task)	Line 195 dispatchSession(sched_task_t *task)

/* receive & decode packet */	/* receive & decode packet */
if ((ret = recv(TASK_FD(task), sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) {	if ((ret = recv(TASK_FD(task), sess->sess_buf->msg_base, sess->sess_buf->msg_len, 0)) == -1) {
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));	EVERBOSE(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
TAILQ_REMOVE(&Sessions, sess, sess_node);	TAILQ_REMOVE(&Sessions, sess, sess_node);
finiSession(sess);	finiSession(sess);
return NULL;	return NULL;
} else if (!ret) {	} else if (!ret) {
ioDEBUG(4, "Session %s EOF received.", sess->sess_cid);	EVERBOSE(4, "Session %s EOF received.", sess->sess_cid);
TAILQ_REMOVE(&Sessions, sess, sess_node);	TAILQ_REMOVE(&Sessions, sess, sess_node);
finiSession(sess);	finiSession(sess);
return NULL;	return NULL;
Line 210 dispatchSession(sched_task_t *task)	Line 210 dispatchSession(sched_task_t *task)
/* dispatch message type */	/* dispatch message type */
if ((len = mqtt_srv_cliDispatch(sess->sess_srv, ret, sess)) < 0) {	if ((len = mqtt_srv_cliDispatch(sess->sess_srv, ret, sess)) < 0) {
if (len == -1) {	if (len == -1) {
ioLIBERR(mqtt);	ELIBERR(mqtt);
finiSession(sess);	finiSession(sess);
} else if (len == -2) {	} else if (len == -2) {
TAILQ_REMOVE(&Sessions, sess, sess_node);	TAILQ_REMOVE(&Sessions, sess, sess_node);
Line 222 dispatchSession(sched_task_t *task)	Line 222 dispatchSession(sched_task_t *task)
} while (len > 0 && ret > 0);	} while (len > 0 && ret > 0);

if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) {	if (len >= 0 && !schedRead(root, dispatchSession, TASK_ARG(task), TASK_FD(task), NULL, 0)) {
ioLIBERR(sched);	ELIBERR(sched);
TAILQ_REMOVE(&Sessions, sess, sess_node);	TAILQ_REMOVE(&Sessions, sess, sess_node);
finiSession(sess);	finiSession(sess);
}	}
Line 239 loadSubscribes(struct tagSession * __restrict sess, mq	Line 239 loadSubscribes(struct tagSession * __restrict sess, mq
return -1;	return -1;

for (i = 0; subs[i].sub_topic.msg_base; i++) {	for (i = 0; subs[i].sub_topic.msg_base; i++) {
store = io_malloc(sizeof(struct tagStore));	store = e_malloc(sizeof(struct tagStore));
if (!store) {	if (!store) {
ioSYSERR(0);	ELIBERR(elwix);
continue;	continue;
} else {	} else {
store->st_msgid = 0;	store->st_msgid = 0;
Line 269 startSession(sched_task_t *task)	Line 269 startSession(sched_task_t *task)
struct timespec ts = { RETAIN_TIMEOUT, 0 };	struct timespec ts = { RETAIN_TIMEOUT, 0 };
intptr_t sock;	intptr_t sock;

ioTRACE(4);	ETRACE();

assert(task);	assert(task);

Line 277 startSession(sched_task_t *task)	Line 277 startSession(sched_task_t *task)
v = TASK_ARG(task);	v = TASK_ARG(task);
/* flow from accept new clients */	/* flow from accept new clients */
sess = initSession(TASK_FD(task), v);	sess = initSession(TASK_FD(task), v);
io_freeVar(&v);	ait_freeVar(&v);
if (!sess) {	if (!sess) {
close(TASK_FD(task));	close(TASK_FD(task));
return NULL;	return NULL;
Line 285 startSession(sched_task_t *task)	Line 285 startSession(sched_task_t *task)

/* receive & decode packet */	/* receive & decode packet */
if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {	if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));	EVERBOSE(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
finiSession(sess);	finiSession(sess);
return NULL;	return NULL;
}	}
Line 303 startSession(sched_task_t *task)	Line 303 startSession(sched_task_t *task)
ret = cack.retcode;	ret = cack.retcode;
flg.flags = cack.reserved;	flg.flags = cack.reserved;
if (flg.reserved) {	if (flg.reserved) {
ioDEBUG(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());	EVERBOSE(3, "Error:: in MQTT protocol #%d - %s", mqtt_GetErrno(), mqtt_GetError());
goto end;	goto end;
} else {	} else {
sess->sess_clean = flg.clean_sess;	sess->sess_clean = flg.clean_sess;
Line 316 startSession(sched_task_t *task)	Line 316 startSession(sched_task_t *task)

/* check online table for user */	/* check online table for user */
if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) {	if (call.LoginACC(&cfg, acc, sess->sess_user, sess->sess_pass) < 1) {
ioDEBUG(0, "Login:: DENIED for username %s and password %s",	EVERBOSE(0, "Login:: DENIED for username %s and password %s",
sess->sess_user, sess->sess_pass);	sess->sess_user, sess->sess_pass);
ret = MQTT_RETCODE_DENIED;	ret = MQTT_RETCODE_DENIED;
goto end;	goto end;
} else {	} else {
ioDEBUG(1, "Login:: ALLOWED for username %s ...", sess->sess_user);	EVERBOSE(1, "Login:: ALLOWED for username %s ...", sess->sess_user);
ret = MQTT_RETCODE_ACCEPTED;	ret = MQTT_RETCODE_ACCEPTED;
}	}

/* db management */	/* db management */
if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {	if (call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%") > 0) {
ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid);	EVERBOSE(2, "Old session %s should be disconnect!", sess->sess_cid);
TAILQ_FOREACH(s, &Sessions, sess_node)	TAILQ_FOREACH(s, &Sessions, sess_node)
if (!strcmp(s->sess_cid, sess->sess_cid)) {	if (!strcmp(s->sess_cid, sess->sess_cid)) {
/* found stale session & disconnect it! */	/* found stale session & disconnect it! */
Line 340 startSession(sched_task_t *task)	Line 340 startSession(sched_task_t *task)
if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr,	if (call.InitSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, sess->sess_addr,
sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg,	sess->sess_will.flag, sess->sess_will.topic, sess->sess_will.msg,
sess->sess_will.qos, sess->sess_will.retain) == -1) {	sess->sess_will.qos, sess->sess_will.retain) == -1) {
ioDEBUG(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user);	EVERBOSE(0, "Session %s DENIED for username %s", sess->sess_cid, sess->sess_user);
ret = MQTT_RETCODE_DENIED;	ret = MQTT_RETCODE_DENIED;
goto end;	goto end;
} else {	} else {
ioDEBUG(0, "Session %s from %s and username %s is started",	EVERBOSE(0, "Session %s from %s and username %s is started",
sess->sess_cid, sess->sess_addr, sess->sess_user);	sess->sess_cid, sess->sess_addr, sess->sess_user);
ret = MQTT_RETCODE_ACCEPTED;	ret = MQTT_RETCODE_ACCEPTED;
}	}
Line 367 startSession(sched_task_t *task)	Line 367 startSession(sched_task_t *task)

/* Start session task OK ... */	/* Start session task OK ... */
if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) {	if (!schedRead(root, dispatchSession, sess, TASK_FD(task), NULL, 0)) {
ioLIBERR(sched);	ELIBERR(sched);
ret = MQTT_RETCODE_DENIED;	ret = MQTT_RETCODE_DENIED;
} else	} else
TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);	TAILQ_INSERT_TAIL(&Sessions, sess, sess_node);
Line 377 startSession(sched_task_t *task)	Line 377 startSession(sched_task_t *task)
end: /* close client connection */	end: /* close client connection */
wlen = mqtt_msgCONNACK(sess->sess_buf, ret);	wlen = mqtt_msgCONNACK(sess->sess_buf, ret);
if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) {	if ((wlen = send(TASK_FD(task), sess->sess_buf->msg_base, wlen, 0)) == -1) {
ioDEBUG(3, "Error:: send(%ld) #%d - %s", (long) TASK_FD(task), errno, strerror(errno));	EVERBOSE(3, "Error:: send(%ld) #%d - %s", (long) TASK_FD(task), errno, strerror(errno));
} else	} else
ioDEBUG(5, "Sended %d bytes", wlen);	EVERBOSE(5, "Sended %d bytes", wlen);

if (ret != MQTT_RETCODE_ACCEPTED) {	if (ret != MQTT_RETCODE_ACCEPTED) {
ioDEBUG(1, "Close client %s with socket=%ld", sess->sess_addr, (long) TASK_FD(task));	EVERBOSE(1, "Close client %s with socket=%ld", sess->sess_addr, (long) TASK_FD(task));
finiSession(sess);	finiSession(sess);
}	}
return NULL;	return NULL;
Line 392 static void *	Line 392 static void *
acceptClient(sched_task_t *task)	acceptClient(sched_task_t *task)
{	{
int cli;	int cli;
io_sockaddr_t sa;	sockaddr_t sa;
socklen_t sslen = sizeof sa.ss;	socklen_t sslen = sizeof sa.ss;
ait_val_t *v;	ait_val_t *v;
char str[STRSIZ];	char str[STRSIZ];

ioTRACE(4);	ETRACE();

assert(task);	assert(task);

Line 406 acceptClient(sched_task_t *task)	Line 406 acceptClient(sched_task_t *task)
else	else
fcntl(cli, F_SETFL, fcntl(cli, F_GETFL) \| O_NONBLOCK);	fcntl(cli, F_SETFL, fcntl(cli, F_GETFL) \| O_NONBLOCK);

v = io_allocVar();	v = ait_allocVar();
if (!v) {	if (!v) {
ioLIBERR(io);	ELIBERR(elwix);
close(cli);	close(cli);
goto end;	goto end;
} else {	} else {
memset(str, 0, sizeof str);	memset(str, 0, sizeof str);
snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));	snprintf(str, sizeof str, "%s:%hu", e_n2addr(&sa, v), e_n2port(&sa));
AIT_FREE_VAL(v);	AIT_FREE_VAL(v);
AIT_SET_STR(v, str);	AIT_SET_STR(v, str);
}	}
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));	EVERBOSE(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));

if (!schedRead(root, startSession, v, cli, NULL, 0)) {	if (!schedRead(root, startSession, v, cli, NULL, 0)) {
io_freeVar(&v);	ait_freeVar(&v);
close(cli);	close(cli);
ioDEBUG(1, "Terminated client with socket=%d", cli);	EVERBOSE(1, "Terminated client with socket=%d", cli);
}	}
end:	end:
if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0))	if (!schedRead(root, acceptClient, NULL, TASK_FD(task), NULL, 0))
ioLIBERR(sched);	ELIBERR(sched);
return NULL;	return NULL;
}	}

Line 438 Run(int sock)	Line 438 Run(int sock)
struct tagSession *sess;	struct tagSession *sess;
struct timespec pl = { 0, 100000000 };	struct timespec pl = { 0, 100000000 };

ioTRACE(1);	ETRACE();

if (mqtt_srv_Listen(sock, 0, 1) == -1) {	if (mqtt_srv_Listen(sock, 0, 1) == -1) {
ioLIBERR(mqtt);	ELIBERR(mqtt);
return -1;	return -1;
}	}

/* state machine - accept new connections */	/* state machine - accept new connections */
if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) {	if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) {
ioLIBERR(sched);	ELIBERR(sched);
return -1;	return -1;
}	}

FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>

Removed from v.1.5
changed lines
	Added in v.1.5.2.2