--- mqtt/src/mqtt_subs.c	2012/06/20 15:01:30	1.2.2.16
+++ mqtt/src/mqtt_subs.c	2012/07/03 08:07:15	1.2.2.19
@@ -126,7 +126,7 @@ execProc(sched_task_t *task)
 	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));
+			(char*) TASK_ARG(task), (u_short) TASK_VAL(task), (u_int) TASK_DATLEN(task));
 	if (TASK_ARG(task))
 		io_free(TASK_ARG(task));
 
@@ -148,6 +148,7 @@ pubRX(sched_task_t *task)
 	void *data = NULL;
 	u_short mid;
 	time_t tim;
+	struct mqtthdr *hdr;
 
 	rlen = RecvFrom(TASK_FD(task));
 	if (rlen == -1)
@@ -156,27 +157,61 @@ pubRX(sched_task_t *task)
 		Kill++;
 		return NULL;
 	}
-	siz = mqtt_readPUBLISH(args->msg, szTopic, sizeof szTopic, &mid, &data);
-	if (siz == -1)
-		goto end;
 
-	/* 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: %04hu, Length: %u, Topic: %s\n", 
-			szTime, mid, siz, szTopic);
+	while (rlen > 0) {
+		hdr = (struct mqtthdr*) args->msg->msg_base;
 
-	if (data) {
-		fputs((const char*) data, TASK_ARG(task));
-		free(data);
-	}
+		switch (hdr->mqtt_msg.type) {
+			case MQTT_TYPE_PUBLISH:
+				siz = mqtt_readPUBLISH(args->msg, szTopic, sizeof szTopic, &mid, &data);
+				if (siz == -1)
+					goto end;
+				else {
+					siz = mqtt_pktLen(hdr);
+					rlen -= siz;
+					ioVERBOSE(4) printf("Remains %d bytes, packet %d bytes\n", 
+							rlen, siz);
+				}
 
-	fprintf(TASK_ARG(task), "\n");
-	fflush(TASK_ARG(task));
+				/* 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: %04hu, QoS: %hhu, "
+						"Length: %u, Topic: %s\n", szTime, mid, hdr->mqtt_msg.qos, 
+						siz, szTopic);
 
-	/* if exists exec script */
-	if (!AIT_ISEMPTY(&args->Value))
-		schedEvent(root, execProc, io_strdup(szTopic), mid, data, siz);
+				if (data) {
+					fputs((const char*) data, TASK_ARG(task));
+					free(data);
+				}
+
+				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);
+
+				memmove(args->msg->msg_base, args->msg->msg_base + siz, rlen);
+				break;
+			case MQTT_TYPE_PINGREQ:
+				siz = mqtt_msgPINGRESP(args->msg);
+				if (siz == -1)
+					goto end;
+				else
+					rlen -= siz;
+
+				/* send ping reply */
+				if (SendTo(TASK_FD(task), siz) == -1)
+					goto end;
+
+				memmove(args->msg->msg_base, args->msg->msg_base + siz, rlen);
+				break;
+			default:
+				ioVERBOSE(1) printf("Unwanted message type #%d ...\n", hdr->mqtt_msg.type);
+				goto end;
+		}
+	}
 end:
 	schedReadSelf(task);
 	return NULL;
@@ -390,18 +425,18 @@ main(int argc, char **argv)
 		else
 			lf = stdout;
 		if (lf) {
-			root = schedBegin();
-
 			ret = Subscribe(args->cli->sock, lf);
 
+			root = schedBegin();
+
 			schedRead(root, pubRX, lf, args->cli->sock, NULL, 0);
 			schedRun(root, &Kill);
 
+			schedEnd(&root);
+
 			if (un)
 				Unsubscribe(args->cli->sock);
 			fclose(lf);
-
-			schedEnd(&root);
 		} else
 			printf("Error:: in output file #%d - %s\n", errno, strerror(errno));
 	} else