mqtt/src/daemon.c - diff

Return to daemon.c CVS log

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

Diff for /mqtt/src/daemon.c between versions 1.2.2.8 and 1.2.2.9

version 1.2.2.8, 2012/04/15 23:45:34	version 1.2.2.9, 2012/04/24 13:54:50
Line 21 initSession(int sock, ait_val_t * __restrict v)	Line 21 initSession(int sock, ait_val_t * __restrict v)

sess = malloc(sizeof(struct tagSession));	sess = malloc(sizeof(struct tagSession));
if (!sess) {	if (!sess) {
ioDEBUG(3, "Error:: in malloc #%d - %s", errno, strerror(errno));	ioSYSERR(0);
io_freeVar(v);	io_freeVar(v);
return NULL;	return NULL;
} else	} else
Line 29 initSession(int sock, ait_val_t * __restrict v)	Line 29 initSession(int sock, ait_val_t * __restrict v)

pthread_mutex_init(&sess->sess_mtx, NULL);	pthread_mutex_init(&sess->sess_mtx, NULL);

SLIST_INIT(&sess->sess_txque[MQTT_QOS_ONCE]);	SLIST_INIT(&sess->sess_subscr);
SLIST_INIT(&sess->sess_txque[MQTT_QOS_ACK]);
SLIST_INIT(&sess->sess_txque[MQTT_QOS_EXACTLY]);

str = cfg_getAttribute(&cfg, "mqttd", "retry");	str = cfg_getAttribute(&cfg, "mqttd", "retry");
if (!str)	if (!str)
Line 39 initSession(int sock, ait_val_t * __restrict v)	Line 37 initSession(int sock, ait_val_t * __restrict v)
else	else
sess->sess_retry = strtol(str, NULL, 0);	sess->sess_retry = strtol(str, NULL, 0);

	if (!(sess->sess_root = schedBegin())) {
	ioLIBERR(sched);
	free(sess);
	io_freeVar(v);
	return NULL;
	}

sess->sess_buf = mqtt_msgAlloc(USHRT_MAX);	sess->sess_buf = mqtt_msgAlloc(USHRT_MAX);
if (!sess->sess_buf) {	if (!sess->sess_buf) {
ioDEBUG(3, "Error:: in msgAlloc #%d - %s", mqtt_GetErrno(), mqtt_GetError());	ioLIBERR(mqtt);
	schedEnd(&sess->sess_root);
free(sess);	free(sess);
io_freeVar(v);	io_freeVar(v);
return NULL;	return NULL;
Line 52 initSession(int sock, ait_val_t * __restrict v)	Line 58 initSession(int sock, ait_val_t * __restrict v)
if (!sess->sess_srv) {	if (!sess->sess_srv) {
ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());	ioDEBUG(3, "Error:: in srv_Init #%d - %s", mqtt_GetErrno(), mqtt_GetError());
mqtt_msgFree(&sess->sess_buf, 42);	mqtt_msgFree(&sess->sess_buf, 42);
	schedEnd(&sess->sess_root);
free(sess);	free(sess);
io_freeVar(v);	io_freeVar(v);
return NULL;	return NULL;
Line 72 initSession(int sock, ait_val_t * __restrict v)	Line 79 initSession(int sock, ait_val_t * __restrict v)
}	}

static void	static void
finiSession(struct tagSession *sess, int preservSock)	finiSession(struct tagSession *sess)
{	{
struct tagStore *store;	struct tagStore *store;
register int i;

ioTRACE(5);	ioTRACE(5);

Line 86 finiSession(struct tagSession *sess, int preservSock)	Line 92 finiSession(struct tagSession *sess, int preservSock)
call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");	call.FiniSessPUB(&cfg, pub, sess->sess_cid, sess->sess_user, "%");

SESS_ELEM_LOCK(sess);	SESS_ELEM_LOCK(sess);
for (i = 0; i < MQTT_QOS_RESERVED; i++)
while ((store = SLIST_FIRST(&sess->sess_txque[i]))) {
SLIST_REMOVE_HEAD(&sess->sess_txque[i], st_node);

if (store->st_subscr.sub_topic.msg_base)	while ((store = SLIST_FIRST(&sess->sess_subscr))) {
free(store->st_subscr.sub_topic.msg_base);	SLIST_REMOVE_HEAD(&sess->sess_subscr, st_node);
if (store->st_subscr.sub_value.msg_base)
free(store->st_subscr.sub_value.msg_base);

free(store);	if (store->st_subscr.sub_topic.msg_base)
}	free(store->st_subscr.sub_topic.msg_base);
	if (store->st_subscr.sub_value.msg_base)
	free(store->st_subscr.sub_value.msg_base);

	free(store);
	}
SESS_ELEM_UNLOCK(sess);	SESS_ELEM_UNLOCK(sess);
pthread_mutex_destroy(&sess->sess_mtx);	pthread_mutex_destroy(&sess->sess_mtx);

	schedEnd(&sess->sess_root);

if (sess->sess_will.msg)	if (sess->sess_will.msg)
free(sess->sess_will.msg);	free(sess->sess_will.msg);
if (sess->sess_will.topic)	if (sess->sess_will.topic)
free(sess->sess_will.topic);	free(sess->sess_will.topic);

if (sess->sess_sock > STDERR_FILENO && !preservSock)	if (sess->sess_sock > STDERR_FILENO)
srv_Close(sess->sess_sock);	srv_Close(sess->sess_sock);

mqtt_srv_Fini(&sess->sess_srv);	mqtt_srv_Fini(&sess->sess_srv);
Line 129 stopSession(struct tagSession *sess)	Line 137 stopSession(struct tagSession *sess)
SESS_UNLOCK;	SESS_UNLOCK;

ret = mqtt_msgDISCONNECT(&msg);	ret = mqtt_msgDISCONNECT(&msg);
if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1)	send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL);
ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));	free(msg.msg_base);
else {
ioDEBUG(5, "Sended %d bytes for disconnect", ret);
free(msg.msg_base);
memset(&msg, 0, sizeof msg);
}

ioDEBUG(1, "Close socket=%d", sess->sess_sock);	ioDEBUG(1, "Close socket=%d", sess->sess_sock);
finiSession(sess, 0);	finiSession(sess);

call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,	call.LOG(logg, "Session %s stopped from %s for user %s.\n", sess->sess_cid,
sess->sess_addr, sess->sess_user);	sess->sess_addr, sess->sess_user);
Line 157 KASession(struct tagSession *sess)	Line 160 KASession(struct tagSession *sess)

/* ping request */	/* ping request */
ret = mqtt_msgPINGREQ(&msg);	ret = mqtt_msgPINGREQ(&msg);
if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {	if ((ret = send(sess->sess_sock, msg.msg_base, ret, MSG_NOSIGNAL)) == -1) {
ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));	ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
return -1;	return -1;
} else {	} else {
Line 239 thrSession(struct tagSession *sess)	Line 242 thrSession(struct tagSession *sess)
continue;	continue;
case MQTT_TYPE_DISCONNECT:	case MQTT_TYPE_DISCONNECT:
ioDEBUG(5, "Exec DISCONNECT session");	ioDEBUG(5, "Exec DISCONNECT session");
finiSession(sess, 0);	finiSession(sess);
locKill ^= locKill;	locKill ^= locKill;
continue;	continue;
case MQTT_TYPE_PUBLISH:	case MQTT_TYPE_PUBLISH:
Line 296 startSession(sched_task_t *task)	Line 299 startSession(sched_task_t *task)
/* receive & decode packet */	/* receive & decode packet */
if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {	if (recv(sess->sess_sock, buf.msg_base, buf.msg_len, 0) == -1) {
ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));	ioDEBUG(3, "Error:: recv(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
finiSession(sess, 0);	finiSession(sess);
return NULL;	return NULL;
}	}
} else {	} else {
Line 354 startSession(sched_task_t *task)	Line 357 startSession(sched_task_t *task)
ret = mqtt_msgCONNACK(&msg, ret);	ret = mqtt_msgCONNACK(&msg, ret);
if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {	if ((ret = send(sess->sess_sock, msg.msg_base, ret, 0)) == -1) {
ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));	ioDEBUG(3, "Error:: send(%d) #%d - %s", sess->sess_sock, errno, strerror(errno));
finiSession(sess, 0);	finiSession(sess);
return NULL;	return NULL;
} else {	} else {
ioDEBUG(5, "Sended %d bytes", ret);	ioDEBUG(5, "Sended %d bytes", ret);
Line 387 end: /* close client connection */	Line 390 end: /* close client connection */
}	}

ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock);	ioDEBUG(1, "Close client %s with socket=%d", sess->sess_addr, sess->sess_sock);
finiSession(sess, 0);	finiSession(sess);
return NULL;	return NULL;
}	}

/* ----------------------------------------------------------------------- */

static void *	static void *
thrSched(void *arg __unused)	acceptClient(sched_task_t *task)
{	{
struct tagSession *sess;	int cli;
struct timespec pl = { 0, 100000000 };	io_sockaddr_t sa;
	socklen_t sslen = sizeof sa.ss;
	ait_val_t *v;
	char str[STRSIZ];

ioTRACE(1);	ioTRACE(4);

/* start scheduler loop */	assert(task);
schedPolling(root, &pl, NULL);
schedRun(root, &Kill);

TAILQ_FOREACH(sess, &Sessions, sess_node)	if ((cli = accept(TASK_FD(task), &sa.sa, &sslen)) == -1)
if (sess->sess_tid)	goto end;
pthread_cancel(sess->sess_tid);	else
ioDEBUG(5, "EXIT from Scheduler thread !!!");	fcntl(TASK_FD(task), F_SETFL, fcntl(TASK_FD(task), F_GETFL, 0) \| O_NONBLOCK);
pthread_exit(NULL);
	v = io_allocVar();
	if (!v) {
	ioLIBERR(io);
	close(cli);
	goto end;
	} else {
	memset(str, 0, sizeof str);
	snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));
	AIT_SET_STR(v, str);
	}
	ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));

	if (!schedRead(root, startSession, v, cli, NULL, 0)) {
	io_freeVar(v);
	close(cli);
	ioDEBUG(1, "Terminated client with socket=%d", cli);
	}
	end:
	schedRead(TASK_ROOT(task), acceptClient, NULL, TASK_FD(task), NULL, 0);
	return NULL;
}	}

	/* ----------------------------------------------------------------------- */

int	int
Run(int sock)	Run(int sock)
{	{
io_sockaddr_t sa;	struct tagPub *pub;
socklen_t sslen = sizeof sa.ss;	struct timespec pl = { 0, 100000000 };
int cli;
pthread_t tid;
ait_val_t *v;
char str[STRSIZ];

ioTRACE(1);	ioTRACE(1);

/* start alternative thread for scheduler */	if (listen(sock, SOMAXCONN) == -1) {
if (pthread_create(&tid, NULL, thrSched, NULL)) {
ioSYSERR(0);	ioSYSERR(0);
return -1;	return -1;
} else	} else
pthread_detach(tid);	fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) \| O_NONBLOCK);
ioDEBUG(2, "Run scheduler management thread");

if (listen(sock, SOMAXCONN) == -1) {	/* state machine - accept new connections */
ioSYSERR(0);	if (!schedRead(root, acceptClient, NULL, sock, NULL, 0)) {
	ioLIBERR(sched);
return -1;	return -1;
}	}

/* state machine - accept new connections */	schedPolling(root, &pl, NULL);
while (!Kill) {	schedRun(root, &Kill);
if ((cli = accept(sock, &sa.sa, &sslen)) == -1) {
if (!Kill)
continue;
ioSYSERR(0);
break;
}
v = io_allocVar();
if (!v) {
ioLIBERR(io);
break;
} else {
memset(str, 0, sizeof str);
snprintf(str, sizeof str, "%s:%hu", io_n2addr(&sa, v), io_n2port(&sa));
AIT_FREE_VAL(v);
AIT_SET_STR(v, str);
}
ioDEBUG(1, "Connected client with socket=%d from %s", cli, AIT_GET_STR(v));

if (!schedRead(root, startSession, v, cli, NULL, 0)) {	/* free all undeleted elements into lists */
io_freeVar(v);	PUBS_LOCK;
close(cli);	TAILQ_FOREACH(pub, &Pubs, pub_node) {
ioDEBUG(1, "Terminated client with socket=%d", cli);	TAILQ_REMOVE(&Pubs, pub, pub_node);
}
}

	AIT_FREE_VAL(&pub->pub_name);
	if (pub->pub_packet.msg_base)
	free(pub->pub_packet.msg_base);
	}
	PUBS_UNLOCK;
return 0;	return 0;
}	}

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

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