mqtt/src/mqtt_subs.c - diff

Return to mqtt_subs.c CVS log

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

Diff for /mqtt/src/mqtt_subs.c between versions 1.2.2.16 and 1.3

version 1.2.2.16, 2012/06/20 15:01:30	version 1.3, 2012/07/03 09:02:50
Line 126 execProc(sched_task_t *task)	Line 126 execProc(sched_task_t *task)
char szLine[MAXPATHLEN + BUFSIZ] = { 0 };	char szLine[MAXPATHLEN + BUFSIZ] = { 0 };

snprintf(szLine, sizeof szLine, "%s '%s' %hu %u", AIT_GET_STR(&args->Value),	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))	if (TASK_ARG(task))
io_free(TASK_ARG(task));	io_free(TASK_ARG(task));

Line 148 pubRX(sched_task_t *task)	Line 148 pubRX(sched_task_t *task)
void *data = NULL;	void *data = NULL;
u_short mid;	u_short mid;
time_t tim;	time_t tim;
	struct mqtthdr *hdr;

rlen = RecvFrom(TASK_FD(task));	rlen = RecvFrom(TASK_FD(task));
if (rlen == -1)	if (rlen == -1)
Line 156 pubRX(sched_task_t *task)	Line 157 pubRX(sched_task_t *task)
Kill++;	Kill++;
return NULL;	return NULL;
}	}
siz = mqtt_readPUBLISH(args->msg, szTopic, sizeof szTopic, &mid, &data);
if (siz == -1)
goto end;

/* send to output */	while (rlen > 0) {
tim = time(NULL);	hdr = (struct mqtthdr*) args->msg->msg_base;
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);

if (data) {	switch (hdr->mqtt_msg.type) {
fputs((const char*) data, TASK_ARG(task));	case MQTT_TYPE_PUBLISH:
free(data);	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");	/* send to output */
fflush(TASK_ARG(task));	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 (data) {
if (!AIT_ISEMPTY(&args->Value))	fputs((const char*) data, TASK_ARG(task));
schedEvent(root, execProc, io_strdup(szTopic), mid, data, siz);	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:	end:
schedReadSelf(task);	schedReadSelf(task);
return NULL;	return NULL;
Line 390 main(int argc, char **argv)	Line 425 main(int argc, char **argv)
else	else
lf = stdout;	lf = stdout;
if (lf) {	if (lf) {
root = schedBegin();

ret = Subscribe(args->cli->sock, lf);	ret = Subscribe(args->cli->sock, lf);

	root = schedBegin();

schedRead(root, pubRX, lf, args->cli->sock, NULL, 0);	schedRead(root, pubRX, lf, args->cli->sock, NULL, 0);
schedRun(root, &Kill);	schedRun(root, &Kill);

	schedEnd(&root);

if (un)	if (un)
Unsubscribe(args->cli->sock);	Unsubscribe(args->cli->sock);
fclose(lf);	fclose(lf);

schedEnd(&root);
} else	} else
printf("Error:: in output file #%d - %s\n", errno, strerror(errno));	printf("Error:: in output file #%d - %s\n", errno, strerror(errno));
} else	} else

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

Removed from v.1.2.2.16
changed lines
	Added in v.1.3