mqtt/src/mqttd.c - diff

Return to mqttd.c CVS log

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

Diff for /mqtt/src/mqttd.c between versions 1.1.1.1.2.3 and 1.2.2.7

version 1.1.1.1.2.3, 2011/11/29 23:23:31	version 1.2.2.7, 2012/05/05 14:51:02
Line 1	Line 1
#include "global.h"	#include "global.h"
	#include "mqttd.h"
#include "rtlm.h"	#include "rtlm.h"
	#include "utils.h"
	#include "daemon.h"


sl_config cfg;	io_enableDEBUG;

	cfg_root_t cfg;
	sessions_t Sessions;
	pubs_t Pubs;
	sched_root_task_t *root;
	sqlite3 acc, pub;
	FILE *logg;
extern char compiled[], compiledby[], compilehost[];	extern char compiled[], compiledby[], compilehost[];
int Verbose;	static char szCfgName[MAXPATHLEN];
	volatile intptr_t Kill;


static void	static void
Line 13 Usage(void)	Line 24 Usage(void)
printf( " -= MQTT Broker =- MQTT Service from ELWIX\n"	printf( " -= MQTT Broker =- MQTT Service from ELWIX\n"
"=== %s@%s === Compiled: %s ===\n\n"	"=== %s@%s === Compiled: %s ===\n\n"
"\t-c <config>\tService config\n"	"\t-c <config>\tService config\n"
	"\t-b\t\tBatch mode\n"
"\t-v\t\tVerbose (more -vvv, more verbose)\n"	"\t-v\t\tVerbose (more -vvv, more verbose)\n"
"\t-h\t\tHelp! This screen\n\n",	"\t-h\t\tHelp! This screen\n\n",
compiledby, compilehost, compiled);	compiledby, compilehost, compiled);
}	}

	static void
	sigHand(int sig)
	{
	int stat;

	switch (sig) {
	case SIGHUP:
	cfgUnloadConfig(&cfg);
	if (!cfgLoadConfig(szCfgName, &cfg)) {
	ioDEBUG(1, "Config reload OK!");
	break;
	}

	ioLIBERR(cfg);
	case SIGTERM:
	ioDEBUG(1, "Terminate MQTT service in progress");
	Kill++;
	break;
	case SIGCHLD:
	while (waitpid(-1, &stat, WNOHANG) > 0);
	break;
	case SIGPIPE:
	break;
	}
	}


int	int
main(int argc, char **argv)	main(int argc, char **argv)
{	{
char ch, szCfgName[MAXPATHLEN];	char ch, batch = 0;
register int i;	register int i;
sqlite3 acc = NULL, pub = NULL;	int sock = -1, ret = 0;
FILE *logg = NULL;	struct passwd *pass;
	struct sigaction sa;
	ait_val_t v;

	TAILQ_INIT(&Sessions);
	TAILQ_INIT(&Pubs);

strlcpy(szCfgName, DEFAULT_CONFIG, sizeof szCfgName);	strlcpy(szCfgName, DEFAULT_CONFIG, sizeof szCfgName);
while ((ch = getopt(argc, argv, "hvc:")) != -1)	while ((ch = getopt(argc, argv, "hvbc:")) != -1)
switch (ch) {	switch (ch) {
case 'c':	case 'c':
strlcpy(szCfgName, optarg, sizeof szCfgName);	strlcpy(szCfgName, optarg, sizeof szCfgName);
break;	break;
	case 'b':
	batch++;
	break;
case 'v':	case 'v':
Verbose++;	io_incDebug;
break;	break;
case 'h':	case 'h':
default:	default:
Line 44 main(int argc, char **argv)	Line 90 main(int argc, char **argv)
argc -= optind;	argc -= optind;
argv += optind;	argv += optind;

if (LoadConfig(szCfgName, &cfg)) {	if (cfgLoadConfig(szCfgName, &cfg)) {
printf("Error:: can't load #%d - %s\n", cfg_GetErrno(), cfg_GetError());	printf("Error:: can't load #%d - %s\n", cfg_GetErrno(), cfg_GetError());
return 1;	return 1;
}	}
	openlog("mqttd", LOG_PID \| LOG_CONS, LOG_DAEMON);
	/* load 3 plugins */
for (i = 0; i < 3; i++)	for (i = 0; i < 3; i++)
if (!mqttLoadRTLM(&cfg, i)) {	if (!mqttLoadRTLM(&cfg, i)) {
printf("Error:: Can't load RTL module\n");	printf("Error:: Can't load RTL module\n");
while (i--)	while (i--)
mqttUnloadRTLM(i);	mqttUnloadRTLM(i);
UnloadConfig(&cfg);	cfgUnloadConfig(&cfg);
	closelog();
return 2;	return 2;
}	}
acc = call.OpenACC(&cfg);	acc = call.OpenACC(&cfg);
if (!acc)	if (!acc) {
	ret = 3;
goto end;	goto end;
	}
pub = call.OpenPUB(&cfg);	pub = call.OpenPUB(&cfg);
if (!pub)	if (!pub) {
	ret = 3;
goto end;	goto end;
	}
logg = call.OpenLOG(&cfg);	logg = call.OpenLOG(&cfg);
if (!logg)	if (!logg) {
	ret = 3;
goto end;	goto end;
	}

if (mqttMkDir(&cfg)) {	if (mqttMkDir(&cfg)) {
printf("Error:: in statedir #%d - %s\n", errno, strerror(errno));	printf("Error:: in statedir #%d - %s\n", errno, strerror(errno));
	ret = 3;
goto end;	goto end;
}	}

VERB(2) printf("Service is ready for start engine ...\n");	if (!batch)
	switch (fork()) {
	case -1:
	printf("Error:: in fork() #%d - %s\n", errno, strerror(errno));
	ret = 5;
	goto end;
	case 0:
	setsid();

end:	ret = open("/dev/null", O_RDWR);
	if (ret != -1) {
	dup2(ret, STDIN_FILENO);
	dup2(ret, STDOUT_FILENO);
	dup2(ret, STDERR_FILENO);
	close(ret);
	}
	ioDEBUG(2, "Welcome MQTT service into shadow land!");
	break;
	default:
	ioDEBUG(2, "MQTT service go to shadow land ...");
	sleep(1);
	ret = 0;
	goto end;
	}
	else
	ioDEBUG(1, "Start service in batch mode ...");

	memset(&sa, 0, sizeof sa);
	sigemptyset(&sa.sa_mask);
	sa.sa_handler = sigHand;
	sigaction(SIGHUP, &sa, NULL);
	sigaction(SIGTERM, &sa, NULL);
	sigaction(SIGCHLD, &sa, NULL);
	sigaction(SIGPIPE, &sa, NULL);
	ioDEBUG(2, "Service is ready for starting engine ...");

	if ((sock = srv_Socket(&cfg)) == -1) {
	ret = 4;
	goto end;
	}

	cfg_loadAttribute(&cfg, "mqttd", "user", &v, MQTT_USER);
	pass = getpwnam(AIT_GET_STR(&v));
	AIT_FREE_VAL(&v);
	if (pass) {
	setgid(pass->pw_gid);
	setuid(pass->pw_uid);
	ioDEBUG(2, "Try to change group #%d and user #%d", pass->pw_gid, pass->pw_uid);
	}

	if (!(root = schedBegin())) {
	ioLIBERR(sched);
	ret = 6;
	goto end;
	}

	/* go catch the cat ... */
	Run(sock);

	schedEnd(&root);
	end: /* free all resources */
	srv_Close(sock);
call.CloseLOG(logg);	call.CloseLOG(logg);
call.ClosePUB(pub);	call.ClosePUB(pub);
call.CloseACC(acc);	call.CloseACC(acc);
for (i = 0; i < 3; i++)	for (i = 0; i < 3; i++)
mqttUnloadRTLM(i);	mqttUnloadRTLM(i);
UnloadConfig(&cfg);	closelog();
return 0;	cfgUnloadConfig(&cfg);
	return ret;
}	}

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

Removed from v.1.1.1.1.2.3
changed lines
	Added in v.1.2.2.7