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.8 and 1.2.2.14

version 1.2.2.8, 2012/05/22 14:15:04	version 1.2.2.14, 2012/06/20 12:43:41
Line 8 io_enableDEBUG;	Line 8 io_enableDEBUG;

extern char compiled[], compiledby[], compilehost[];	extern char compiled[], compiledby[], compilehost[];
volatile intptr_t Kill;	volatile intptr_t Kill;
	sched_root_task_t *root;

struct tagArgs *args;	struct tagArgs *args;

Line 56 Subscribe(int sock, FILE *lf)	Line 57 Subscribe(int sock, FILE *lf)
u_short mid;	u_short mid;
mqtt_subscr_t *sub;	mqtt_subscr_t *sub;

	#ifdef __NetBSD__
	srandom(getpid() ^ time(NULL));
	#else
srandomdev();	srandomdev();
	#endif
mid = random() % USHRT_MAX;	mid = random() % USHRT_MAX;

printf(" > Execute SUBSCRIBE request #%d ... ", mid);	printf(" > Execute SUBSCRIBE request #%d ... ", mid);
Line 80 Unsubscribe(int sock)	Line 85 Unsubscribe(int sock)
{	{
u_short mid;	u_short mid;

	#ifdef __NetBSD__
	srandom(getpid() ^ time(NULL));
	#else
srandomdev();	srandomdev();
	#endif
mid = random() % USHRT_MAX;	mid = random() % USHRT_MAX;

printf(" > Execute UNSUBSCRIBE request #%d ... ", mid);	printf(" > Execute UNSUBSCRIBE request #%d ... ", mid);
Line 93 Unsubscribe(int sock)	Line 102 Unsubscribe(int sock)
return 0;	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 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;

	/* 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");
	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;
	}


int	int
main(int argc, char **argv)	main(int argc, char **argv)
{	{
Line 104 main(int argc, char **argv)	Line 193 main(int argc, char **argv)
int ret = 0;	int ret = 0;
char *str, szStr[STRSIZ], szLogName[MAXPATHLEN] = { 0 };	char *str, szStr[STRSIZ], szLogName[MAXPATHLEN] = { 0 };
FILE *lf;	FILE *lf;
	struct sigaction sa;

if (!(args = malloc(sizeof(struct tagArgs)))) {	if (!(args = io_malloc(sizeof(struct tagArgs)))) {
printf("Error:: in arguments #%d - %s\n", errno, strerror(errno));	printf("Error:: in arguments #%d - %s\n", errno, strerror(errno));
return 1;	return 1;
} else	} else
memset(args, 0, sizeof(struct tagArgs));	memset(args, 0, sizeof(struct tagArgs));
if (!(args->subscr = mqtt_subAlloc(idx))) {	if (!(args->subscr = mqtt_subAlloc(idx))) {
printf("Error:: in subscribes array #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());	printf("Error:: in subscribes array #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
free(args);	io_free(args);
return 1;	return 1;
} else	} else
args->free = cleanArgs;	args->free = cleanArgs;
Line 120 main(int argc, char **argv)	Line 210 main(int argc, char **argv)
if (!(args->msg = mqtt_msgAlloc(USHRT_MAX))) {	if (!(args->msg = mqtt_msgAlloc(USHRT_MAX))) {
printf("Error:: in mqtt buffer #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());	printf("Error:: in mqtt buffer #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
args->free(args);	args->free(args);
free(args);	io_free(args);
return 1;	return 1;
}	}

Line 158 main(int argc, char **argv)	Line 248 main(int argc, char **argv)
if (!sub) {	if (!sub) {
printf("Error:: #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());	printf("Error:: #%d - %s\n", mqtt_GetErrno(), mqtt_GetError());
args->free(args);	args->free(args);
free(args);	io_free(args);
return 1;	return 1;
} else	} else
sub += idx++;	sub += idx++;
Line 181 main(int argc, char **argv)	Line 271 main(int argc, char **argv)
if (args->QoS > MQTT_QOS_EXACTLY) {	if (args->QoS > MQTT_QOS_EXACTLY) {
printf("Error:: invalid QoS level %d\n", args->QoS);	printf("Error:: invalid QoS level %d\n", args->QoS);
args->free(args);	args->free(args);
free(args);	io_free(args);
return 1;	return 1;
}	}
break;	break;
Line 206 main(int argc, char **argv)	Line 296 main(int argc, char **argv)
case 'h':	case 'h':
default:	default:
args->free(args);	args->free(args);
free(args);	io_free(args);
Usage();	Usage();
return 1;	return 1;
}	}
Line 215 main(int argc, char **argv)	Line 305 main(int argc, char **argv)
if (argc < 2) {	if (argc < 2) {
printf("Error:: host for connect not found, connection id or topic not supplied!\n\n");	printf("Error:: host for connect not found, connection id or topic not supplied!\n\n");
args->free(args);	args->free(args);
free(args);	io_free(args);
Usage();	Usage();
return 1;	return 1;
} else {	} else {
Line 229 main(int argc, char **argv)	Line 319 main(int argc, char **argv)
if (!io_gethostbyname(*argv, port, &args->addr)) {	if (!io_gethostbyname(*argv, port, &args->addr)) {
printf("Error:: host not valid #%d - %s\n", io_GetErrno(), io_GetError());	printf("Error:: host not valid #%d - %s\n", io_GetErrno(), io_GetError());
args->free(args);	args->free(args);
free(args);	io_free(args);
Usage();	Usage();
return 1;	return 1;
}	}
printf("Connecting to %s:%d ... ", io_n2addr(&args->addr, &val), io_n2port(&args->addr));	printf("Connecting to %s:%d ... ", io_n2addr(&args->addr, &val), io_n2port(&args->addr));
AIT_FREE_VAL(&val);	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))) {	if (!(args->cli = mqtt_cli_Open(&args->addr.sa, args->ka))) {
args->free(args);	ret = 2;
free(args);	goto end;
return 2;
}	}

switch ((ret = ConnectClient(args->cli->sock))) {	switch ((ret = ConnectClient(args->cli->sock))) {
Line 272 main(int argc, char **argv)	Line 390 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);

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

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

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

mqtt_cli_Close(&args->cli);	mqtt_cli_Close(&args->cli);
	end:
args->free(args);	args->free(args);
free(args);	io_free(args);
return ret;	return ret;
}	}

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

Removed from v.1.2.2.8
changed lines
	Added in v.1.2.2.14