--- mqtt/src/mqtt_subs.c	2012/06/20 09:23:28	1.2.2.13
+++ mqtt/src/mqtt_subs.c	2012/06/20 12:43:41	1.2.2.14
@@ -102,32 +102,81 @@ Unsubscribe(int sock)
 	return 0;
 }
 
+static void
+sigz(int sig)
+{
+	int stat;
 
+	switch (sig) {
+		case SIGINT:
+		case SIGTERM:
+			Kill++;
+			break;
+		case SIGCHLD:
+			while (waitpid(-1, &stat, WNOHANG) > 0);
+			break;
+	}
+}
+
+
 static void *
+execProc(sched_task_t *task)
+{
+	FILE *f;
+	char szLine[MAXPATHLEN + BUFSIZ] = { 0 };
+
+	snprintf(szLine, sizeof szLine, "%s '%s' %hu %u", AIT_GET_STR(&args->Value), 
+			(char*) TASK_ARG(task), (u_short) TASK_VAL(task), TASK_DATLEN(task));
+	if (TASK_ARG(task))
+		io_free(TASK_ARG(task));
+
+	f = popen(szLine, "w");
+	if (!f) {
+		ioSYSERR(0);
+		return NULL;
+	} else
+		fputs(TASK_DATA(task), f);
+	pclose(f);
+	return NULL;
+}
+
+static void *
 pubRX(sched_task_t *task)
 {
 	int siz, rlen;
-	char szTopic[STRSIZ] = { 0 };
+	char szTime[STRSIZ] = { 0 }, szTopic[STRSIZ] = { 0 };
 	void *data = NULL;
 	u_short mid;
+	time_t tim;
 
 	rlen = RecvFrom(TASK_FD(task));
 	if (rlen == -1)
 		goto end;
+	if (!rlen) {
+		Kill++;
+		return NULL;
+	}
 	siz = mqtt_readPUBLISH(args->msg, szTopic, sizeof szTopic, &mid, &data);
 	if (siz == -1)
 		goto end;
 
-	fprintf(TASK_ARG(task), "\nMessage ID: 0x%04hu, Length: %u, Topic: %s\n", 
-			mid, siz, szTopic);
+	/* send to output */
+	tim = time(NULL);
+	strftime(szTime, sizeof szTime, "%Y-%m-%d %H:%M:%S", localtime(&tim));
+	fprintf(TASK_ARG(task), "\n[%s] Message ID: 0x%04hu, Length: %u, Topic: %s\n", 
+			szTime, mid, siz, szTopic);
 
 	if (data) {
 		fputs((const char*) data, TASK_ARG(task));
 		free(data);
 	}
 
-	fprintf(TASK_ARG(task), "\n.\n");
+	fprintf(TASK_ARG(task), "\n");
 	fflush(TASK_ARG(task));
+
+	/* if exists exec script */
+	if (!AIT_ISEMPTY(&args->Value))
+		schedEvent(root, execProc, io_strdup(szTopic), mid, data, siz);
 end:
 	schedReadSelf(task);
 	return NULL;
@@ -144,6 +193,7 @@ main(int argc, char **argv)
 	int ret = 0;
 	char *str, szStr[STRSIZ], szLogName[MAXPATHLEN] = { 0 };
 	FILE *lf;
+	struct sigaction sa;
 
 	if (!(args = io_malloc(sizeof(struct tagArgs)))) {
 		printf("Error:: in arguments #%d - %s\n", errno, strerror(errno));
@@ -276,10 +326,38 @@ main(int argc, char **argv)
 	printf("Connecting to %s:%d ... ", io_n2addr(&args->addr, &val), io_n2port(&args->addr));
 	AIT_FREE_VAL(&val);
 
+	sa.sa_handler = sigz;
+	sigemptyset(&sa.sa_mask);
+	sigaction(SIGTERM, &sa, NULL);
+	sigaction(SIGINT, &sa, NULL);
+	sigaction(SIGCHLD, &sa, NULL);
+
+	if (!batch)
+		switch (fork()) {
+			case -1:	/* error */
+				printf("Error:: in fork() #%d - %s\n", errno, strerror(errno));
+				ret = 2;
+				goto end;
+			case 0:		/* child */
+				setsid();
+
+				ret = open("/dev/null", O_RDWR);
+				if (ret != -1) {
+					dup2(ret, STDIN_FILENO);
+					dup2(ret, STDOUT_FILENO);
+					dup2(ret, STDERR_FILENO);
+					close(ret);
+				}
+				break;
+			default:	/* parent */
+				printf(">> Service started\n");
+				ret = 0;
+				goto end;
+		}
+
 	if (!(args->cli = mqtt_cli_Open(&args->addr.sa, args->ka))) {
-		args->free(args);
-		io_free(args);
-		return 2;
+		ret = 2;
+		goto end;
 	}
 
 	switch ((ret = ConnectClient(args->cli->sock))) {
@@ -325,12 +403,12 @@ main(int argc, char **argv)
 
 			schedEnd(&root);
 		} else
-			printf("Error:: in subscribe file #%d - %s\n", errno, strerror(errno));
+			printf("Error:: in output file #%d - %s\n", errno, strerror(errno));
 	} else
 		ret = 3;
 
 	mqtt_cli_Close(&args->cli);
-
+end:
 	args->free(args);
 	io_free(args);
 	return ret;