--- mqtt/src/daemon.c	2012/07/03 07:48:49	1.2.2.44
+++ mqtt/src/daemon.c	2012/08/29 14:40:08	1.4.2.1
@@ -1,3 +1,48 @@
+/*************************************************************************
+* (C) 2011 AITNET ltd - Sofia/Bulgaria - <misho@aitbg.com>
+*  by Michael Pounov <misho@openbsd-bg.org>
+*
+* $Author: misho $
+* $Id: daemon.c,v 1.4.2.1 2012/08/29 14:40:08 misho Exp $
+*
+**************************************************************************
+The ELWIX and AITNET software is distributed under the following
+terms:
+
+All of the documentation and software included in the ELWIX and AITNET
+Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
+
+Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012
+	by Michael Pounov <misho@elwix.org>.  All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+1. Redistributions of source code must retain the above copyright
+   notice, this list of conditions and the following disclaimer.
+2. Redistributions in binary form must reproduce the above copyright
+   notice, this list of conditions and the following disclaimer in the
+   documentation and/or other materials provided with the distribution.
+3. All advertising materials mentioning features or use of this software
+   must display the following acknowledgement:
+This product includes software developed by Michael Pounov <misho@elwix.org>
+ELWIX - Embedded LightWeight unIX and its contributors.
+4. Neither the name of AITNET nor the names of its contributors
+   may be used to endorse or promote products derived from this software
+   without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+SUCH DAMAGE.
+*/
 #include "global.h"
 #include "rtlm.h"
 #include "mqttd.h"
@@ -77,8 +122,8 @@ finiSession(struct tagSession *sess)
 			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);
+		if (call.WipePUB_topic)	/* only dynamic messages */
+			call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, 0);
 	}
 
 	while ((store = SLIST_FIRST(&sess->sess_subscr))) {
@@ -222,6 +267,7 @@ startSession(sched_task_t *task)
 	int ret, wlen;
 	mqtt_subscr_t *subs;
 	struct timespec ts = { RETAIN_TIMEOUT, 0 };
+	intptr_t sock;
 
 	ioTRACE(4);
 
@@ -285,7 +331,8 @@ startSession(sched_task_t *task)
 		TAILQ_FOREACH(s, &Sessions, sess_node)
 			if (!strcmp(s->sess_cid, sess->sess_cid)) {
 				/* found stale session & disconnect it! */
-				schedCancelby(root, taskMAX, CRITERIA_FD, (void*) s->sess_sock, NULL);
+				sock = s->sess_sock;
+				schedCancelby(root, taskMAX, CRITERIA_FD, (void*) sock, NULL);
 				schedWrite(root, leaveClient, s, s->sess_sock, NULL, 0);
 				break;
 			}
@@ -305,7 +352,7 @@ startSession(sched_task_t *task)
 	if (sess->sess_clean) {
 		if (call.DeletePUB_subscribe)
 			call.DeletePUB_subscribe(&cfg, pub, sess->sess_cid, "%", sess->sess_user, "%");
-		if (call.WipePUB_topic)
+		if (call.WipePUB_topic)	/* delete ALL messages */
 			call.WipePUB_topic(&cfg, pub, sess->sess_cid, sess->sess_user, -1);
 	} else if (call.ReadPUB_subscribe) {
 		/* load subscribes */
@@ -356,6 +403,8 @@ acceptClient(sched_task_t *task)
 
 	if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1)
 		goto end;
+	else
+		fcntl(cli, F_SETFL, fcntl(cli, F_GETFL) | O_NONBLOCK);
 
 	v = io_allocVar();
 	if (!v) {
@@ -404,6 +453,8 @@ Run(int sock)
 
 	schedPolling(root, &pl, NULL);
 	schedRun(root, &Kill);
+
+	schedCancelby(root, taskTIMER, CRITERIA_CALL, sendRetain, NULL);
 
 	/* free all undeleted elements into lists */
 	TAILQ_FOREACH(sess, &Sessions, sess_node) {