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.2.2.27 and 1.2.2.32

version 1.2.2.27, 2012/05/05 15:16:34	version 1.2.2.32, 2012/05/22 14:15:04
Line 39 initSession(int sock, ait_val_t * __restrict v)	Line 39 initSession(int sock, ait_val_t * __restrict v)
}	}

/* init server actor */	/* init server actor */
sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf);	sess->sess_srv = mqtt_srv_Init(sock, sess->sess_buf, sess->sess_ka);
if (!sess->sess_srv) {	if (!sess->sess_srv) {
ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());	ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
mqtt_msgFree(&sess->sess_buf, 42);	mqtt_msgFree(&sess->sess_buf, 42);
Line 70 finiSession(struct tagSession *sess)	Line 70 finiSession(struct tagSession *sess)
if (!sess)	if (!sess)
return;	return;

if (call.FiniSessPUB)	if (sess->sess_clean) {
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");	if (call.FiniSessPUB)
	call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");
	if (call.DeletePUB_subscribe)
	call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%");
	if (call.WipePUB_topic)
	call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1);
	}

while ((store = SLIST_FIRST(&sess->sess_subscr))) {	while ((store = SLIST_FIRST(&sess->sess_subscr))) {
SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);	SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);
Line 180 startSession(sched_task_t *task)	Line 186 startSession(sched_task_t *task)
mqtt_msg_t buf = { basebuf, sizeof basebuf };	mqtt_msg_t buf = { basebuf, sizeof basebuf };
mqtthdr_connflgs_t flg;	mqtthdr_connflgs_t flg;
mqtthdr_connack_t cack;	mqtthdr_connack_t cack;
	ait_val_t *v;
struct tagSession s, sess = NULL;	struct tagSession s, sess = NULL;
int ret, wlen;	int ret, wlen;

Line 188 startSession(sched_task_t *task)	Line 195 startSession(sched_task_t *task)
assert(task);	assert(task);

if (!TASK_DATA(task)) {	if (!TASK_DATA(task)) {
	v = TASK_ARG(task);
/* flow from accept new clients */	/* flow from accept new clients */
sess = initSession(TASK_FD(task), TASK_ARG(task));	sess = initSession(TASK_FD(task), v);
io_freeVar(TASK_ARG(task));	io_freeVar(&v);
if (!sess) {	if (!sess) {
close(TASK_FD(task));	close(TASK_FD(task));
return NULL;	return NULL;
Line 223 startSession(sched_task_t *task)	Line 231 startSession(sched_task_t *task)
sess->sess_will.qos = flg.will_qos;	sess->sess_will.qos = flg.will_qos;
sess->sess_will.retain = flg.will_retain;	sess->sess_will.retain = flg.will_retain;
sess->sess_will.flag = flg.will_flg;	sess->sess_will.flag = flg.will_flg;

	sess->sess_srv->timeout = sess->sess_ka;
}	}

/* check online table for user */	/* check online table for user */
Line 236 startSession(sched_task_t *task)	Line 246 startSession(sched_task_t *task)
ret = MQTT_RETCODE_ACCEPTED;	ret = MQTT_RETCODE_ACCEPTED;
}	}

	/* 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);	ioDEBUG(2, "Old session %s should be disconnect!", sess->sess_cid);
TAILQ_FOREACH(s, &Sessions, sess_node)	TAILQ_FOREACH(s, &Sessions, sess_node)
Line 245 startSession(sched_task_t *task)	Line 256 startSession(sched_task_t *task)
break;	break;
}	}
}	}

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) {
Line 257 startSession(sched_task_t *task)	Line 267 startSession(sched_task_t *task)
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;
}	}
	/* clean/load session if requested */
	if (sess->sess_clean) {
	if (call.DeletePUB_subscribe)
	call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%");
	if (call.WipePUB_topic)
	call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1);
	} else {
	// todo: read_sql subs and prepare publish
	}

/* 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)) {
Line 313 acceptClient(sched_task_t *task)	Line 332 acceptClient(sched_task_t *task)
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));	ioDEBUG(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);	io_freeVar(&v);
close(cli);	close(cli);
ioDEBUG(1, "Terminated client with socket=%d", cli);	ioDEBUG(1, "Terminated client with socket=%d", cli);
}	}
Line 328 end:	Line 347 end:
int	int
Run(int sock)	Run(int sock)
{	{
struct tagPub *pub;
struct tagSession *sess;	struct tagSession *sess;
struct timespec pl = { 0, 100000000 };	struct timespec pl = { 0, 100000000 };

Line 350 Run(int sock)	Line 368 Run(int sock)
schedRun(root, &Kill);	schedRun(root, &Kill);

/* free all undeleted elements into lists */	/* free all undeleted elements into lists */
TAILQ_FOREACH(pub, &Pubs, pub_node) {
TAILQ_REMOVE(&Pubs, pub, pub_node);

AIT_FREE_VAL(&pub->pub_name);
if (pub->pub_packet.msg_base)
free(pub->pub_packet.msg_base);
}
TAILQ_FOREACH(sess, &Sessions, sess_node) {	TAILQ_FOREACH(sess, &Sessions, sess_node) {
TAILQ_REMOVE(&Sessions, sess, sess_node);	TAILQ_REMOVE(&Sessions, sess, sess_node);

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

Removed from v.1.2.2.27
changed lines
	Added in v.1.2.2.32