rem: } while(xj_manage_jab(recv_buff, nr, &pos, jwl->contact_h, jwl->aliases, jcp->ojc[i]) == 9	&& main_loop);
context: int xj_worker_process(xj_wlist jwl, char* jaddress, int jport, int rank,		db_con_t* db_con){	int pipe, ret, i, pos, maxfd, flag;	xj_jcon_pool jcp;	struct timeval tmv;	fd_set set, mset;	xj_sipmsg jsmsg;	str sto;	xj_jcon jbc = NULL;	xj_jconf jcf = NULL;	char *p, buff[1024], recv_buff[4096];	int flags, nr, ltime = 0;	xj_pres_cell prc = NULL;		db_key_t keys[] = {"sip_id", "type"};	db_val_t vals[] = { {DB_STRING, 0, {.string_val = buff}},						{DB_INT, 0, {.int_val = 0}} };	db_key_t col[] = {"jab_id", "jab_passwd"};	db_res_t* res = NULL;	_xj_pid = getpid();		//signal(SIGTERM, xj_sig_handler);	//signal(SIGINT, xj_sig_handler);	//signal(SIGQUIT, xj_sig_handler);	signal(SIGSEGV, xj_sig_handler);	if(!jwl || !jwl->aliases || !jwl->aliases->jdm 			|| !jaddress || rank >= jwl->len)	{		DBG("XJAB:xj_worker[%d]:%d: exiting - wrong parameters\n",				rank, _xj_pid);		return -1;	}	pipe = jwl->workers[rank].rpipe;	DBG("XJAB:xj_worker[%d]:%d: started - pipe=<%d> : 1st message delay"		" <%d>\n", rank, _xj_pid, pipe, jwl->delayt);	if((jcp=xj_jcon_pool_init(jwl->maxj,XJ_MSG_POOL_SIZE,jwl->delayt))==NULL)	{		DBG("XJAB:xj_worker: cannot allocate the pool\n");		return -1;	}	maxfd = pipe;	tmv.tv_sec = jwl->sleept;	tmv.tv_usec = 0;	FD_ZERO(&set);	FD_SET(pipe, &set);	while(main_loop)	{		mset = set;		tmv.tv_sec = (jcp->jmqueue.size == 0)?jwl->sleept:1;		//DBG("XJAB:xj_worker[%d]:%d: select waiting %ds - queue=%d\n",rank,		//		_xj_pid, (int)tmv.tv_sec, jcp->jmqueue.size);		tmv.tv_usec = 0;		ret = select(maxfd+1, &mset, NULL, NULL, &tmv);		/** check the queue AND conecction of head element is ready */		for(i = 0; i<jcp->jmqueue.size && main_loop; i++)		{			if(jcp->jmqueue.jsm[i]==NULL || jcp->jmqueue.ojc[i]==NULL)			{				if(jcp->jmqueue.jsm[i]!=NULL)				{					xj_sipmsg_free(jcp->jmqueue.jsm[i]);					jcp->jmqueue.jsm[i] = NULL;					xj_jcon_pool_del_jmsg(jcp, i);				}				if(jcp->jmqueue.ojc[i]!=NULL)					xj_jcon_pool_del_jmsg(jcp, i);				continue;			}			if(jcp->jmqueue.expire[i] < get_ticks())			{				DBG("XJAB:xj_worker:%d: message to %.*s is expired\n",					_xj_pid, jcp->jmqueue.jsm[i]->to.len, 					jcp->jmqueue.jsm[i]->to.s);				xj_send_sip_msgz(jcp->jmqueue.jsm[i]->jkey->id, 						&jcp->jmqueue.jsm[i]->to, jwl->contact_h, 						XJ_DMSG_ERR_SENDIM, &jcp->jmqueue.ojc[i]->jkey->flag);				if(jcp->jmqueue.jsm[i]!=NULL)				{					xj_sipmsg_free(jcp->jmqueue.jsm[i]);					jcp->jmqueue.jsm[i] = NULL;				}				/** delete message from queue */				xj_jcon_pool_del_jmsg(jcp, i);				continue;			}			DBG("XJAB:xj_worker:%d:%d: QUEUE: message[%d] from [%.*s]/to [%.*s]/"					"body[%.*s] expires at %d\n",					_xj_pid, get_ticks(), i, 					jcp->jmqueue.jsm[i]->jkey->id->len,					jcp->jmqueue.jsm[i]->jkey->id->s,					jcp->jmqueue.jsm[i]->to.len,jcp->jmqueue.jsm[i]->to.s,					jcp->jmqueue.jsm[i]->msg.len,jcp->jmqueue.jsm[i]->msg.s,					jcp->jmqueue.expire[i]);			if(xj_jcon_is_ready(jcp->jmqueue.ojc[i], jcp->jmqueue.jsm[i]->to.s,					jcp->jmqueue.jsm[i]->to.len, jwl->aliases->dlm))				continue;			/*** address corection ***/			flag = XJ_ADDRTR_A2B;			if(!xj_jconf_check_addr(&jcp->jmqueue.jsm[i]->to,jwl->aliases->dlm))				flag |= XJ_ADDRTR_CON;						sto.s = buff; 			sto.len = 0;			if(xj_address_translation(&jcp->jmqueue.jsm[i]->to,				&sto, jwl->aliases, flag) == 0)			{						/** send message from queue */				DBG("XJAB:xj_worker:%d: SENDING THE MESSAGE FROM "					" LOCAL QUEUE TO JABBER NETWORK ...\n", _xj_pid);				xj_jcon_send_msg(jcp->jmqueue.ojc[i],					sto.s, sto.len,					jcp->jmqueue.jsm[i]->msg.s,					jcp->jmqueue.jsm[i]->msg.len,					(flag&XJ_ADDRTR_CON)?XJ_JMSG_GROUPCHAT:XJ_JMSG_CHAT);			}			else				DBG("XJAB:xj_worker:%d: ERROR SENDING THE MESSAGE FROM "				" LOCAL QUEUE TO JABBER NETWORK ...\n", _xj_pid);							if(jcp->jmqueue.jsm[i]!=NULL)			{				xj_sipmsg_free(jcp->jmqueue.jsm[i]);				jcp->jmqueue.jsm[i] = NULL;			}			/** delete message from queue */			xj_jcon_pool_del_jmsg(jcp, i);		} // end MSG queue checking				if(ret <= 0)			goto step_x;				DBG("XJAB:xj_worker:%d: something is coming\n", _xj_pid);		if(!FD_ISSET(pipe, &mset))			goto step_y;				if(read(pipe, &jsmsg, sizeof(jsmsg)) < sizeof(jsmsg))		{			DBG("XJAB:xj_worker:%d: BROKEN PIPE - exiting\n", _xj_pid);			break;		}		DBG("XJAB:xj_worker:%d: job <%p> from SER\n", _xj_pid, jsmsg);				if(jsmsg == NULL || jsmsg->jkey==NULL || jsmsg->jkey->id==NULL)			goto step_w;		strncpy(buff, jsmsg->jkey->id->s, jsmsg->jkey->id->len);		buff[jsmsg->jkey->id->len] = 0;		jbc = xj_jcon_pool_get(jcp, jsmsg->jkey);				switch(jsmsg->type)		{			case XJ_SEND_MESSAGE:				if(!xj_jconf_check_addr(&jsmsg->to, jwl->aliases->dlm) &&				(!jbc||!xj_jcon_get_jconf(jbc,&jsmsg->to,jwl->aliases->dlm)))				{					xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,						jwl->contact_h, XJ_DMSG_ERR_NOTJCONF, NULL);					goto step_w;				}				break;			case XJ_REG_WATCHER:			case XJ_JOIN_JCONF:			case XJ_GO_ONLINE:				break;			case XJ_EXIT_JCONF:				if(jbc == NULL)					goto step_w;				// close the conference session here				if(jbc->nrjconf <= 0)					goto step_w;				if(!xj_jconf_check_addr(&jsmsg->to, jwl->aliases->dlm))					xj_jcon_del_jconf(jbc, &jsmsg->to, jwl->aliases->dlm,						XJ_JCMD_UNSUBSCRIBE);				xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,					jwl->contact_h, XJ_DMSG_INF_JCONFEXIT, NULL);				goto step_w;			case XJ_GO_OFFLINE:				if(jbc != NULL)					jbc->expire = ltime = -1;				goto step_w;			case XJ_DEL_WATCHER:			default:				goto step_w;		}				if(jbc != NULL)		{			DBG("XJAB:xj_worker:%d: connection already exists"				" for <%s> ...\n", _xj_pid, buff);			xj_jcon_update(jbc, jwl->cachet);			goto step_z;		}				// NO OPEN CONNECTION FOR THIS SIP ID		DBG("XJAB:xj_worker:%d: new connection for <%s>.\n",			_xj_pid, buff);				if(db_query(db_con, keys, 0, vals, col, 2, 2, NULL, &res) != 0 ||			RES_ROW_N(res) <= 0)		{			DBG("XJAB:xj_worker:%d: no database result\n", _xj_pid);			xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to, 				jwl->contact_h, XJ_DMSG_ERR_JGWFORB, NULL);						goto step_v;		}				jbc = xj_jcon_init(jaddress, jport);				if(xj_jcon_connect(jbc))		{			DBG("XJAB:xj_worker:%d: Cannot connect"				" to the Jabber server ...\n", _xj_pid);			xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to, 				jwl->contact_h, XJ_DMSG_ERR_NOJSRV, NULL);			goto step_v;		}				DBG("XJAB:xj_worker: auth to jabber as: [%s] / [%s]\n",			(char*)(ROW_VALUES(RES_ROWS(res))[0].val.string_val), 			(char*)(ROW_VALUES(RES_ROWS(res))[1].val.string_val));				if(xj_jcon_user_auth(jbc,			(char*)(ROW_VALUES(RES_ROWS(res))[0].val.string_val),			(char*)(ROW_VALUES(RES_ROWS(res))[1].val.string_val),			XJAB_RESOURCE) < 0)		{			DBG("XJAB:xj_worker:%d: Authentication to the Jabber server"				" failed ...\n", _xj_pid);			xj_jcon_disconnect(jbc);						xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,				jwl->contact_h, XJ_DMSG_ERR_JAUTH, NULL);						xj_jcon_free(jbc);			goto step_v;		}				if(xj_jcon_set_attrs(jbc, jsmsg->jkey, jwl->cachet, jwl->delayt)			|| xj_jcon_pool_add(jcp, jbc))		{			DBG("XJAB:xj_worker:%d: Keeping connection to Jabber server"				" failed! Not enough memory ...\n", _xj_pid);			xj_jcon_disconnect(jbc);			xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to, jwl->contact_h,					XJ_DMSG_ERR_JGWFULL, NULL);			xj_jcon_free(jbc);			goto step_v;		}										/** add socket descriptor to select */		DBG("XJAB:xj_worker:%d: add connection on <%d> \n", _xj_pid, jbc->sock);		if(jbc->sock > maxfd)			maxfd = jbc->sock;		FD_SET(jbc->sock, &set);												xj_jcon_get_roster(jbc);		xj_jcon_send_presence(jbc, NULL, NULL, "Online", "9");				/** wait for a while - the worker is tired */		//sleep(3);				if ((res != NULL) && (db_free_query(db_con,res) < 0))		{			DBG("XJAB:xj_worker:%d:Error while freeing"				" SQL result - worker terminated\n", _xj_pid);			return -1;		}		else			res = NULL;		if(jsmsg->type == XJ_GO_ONLINE)			goto step_w;		step_z:		if(jsmsg->type == XJ_REG_WATCHER)		{ // register a presence watcher			if(!xj_jconf_check_addr(&jsmsg->to, jwl->aliases->dlm))			{ // is for a conference - ignore?!?!				DBG("XJAB:xj_worker:%d: presence request for a conference.\n",					_xj_pid);				// set as offline				(*(jsmsg->cbf))(&jsmsg->to, 0, jsmsg->p);				goto step_w;			}						sto.s = buff; 			sto.len = 0;			if(xj_address_translation(&jsmsg->to, &sto, jwl->aliases, 					XJ_ADDRTR_A2B) == 0)			{				prc = xj_pres_list_check(jbc->plist, &sto);				if(!prc)				{					prc = xj_pres_cell_new();					if(!prc)					{						DBG("XJAB:xj_worker:%d: cannot create a presence"							" cell for %.*s.\n", _xj_pid, sto.len, sto.s);						goto step_w;					}					if(xj_pres_cell_init(prc, &sto, jsmsg->cbf, jsmsg->p)<0)					{						xj_pres_cell_free(prc);						goto step_w;					}					if((prc = xj_pres_list_add(jbc->plist, prc))==NULL)					{						DBG("XJAB:xj_worker:%d: cannot add the presence"							" cell for %.*s.\n", _xj_pid, sto.len, sto.s);						goto step_w;					}					sto.s[sto.len] = 0;					if(!xj_jcon_send_subscribe(jbc, sto.s, NULL, "subscribe"))						prc->status = XJ_PRES_STATUS_WAIT; 				}				else					xj_pres_cell_update(prc, jsmsg->cbf, jsmsg->p);				// send presence info to SIP subscriber				(*(prc->cbf))(&jsmsg->to, prc->state, prc->cbp);			}			goto step_w;		}		flag = 0;		if(!xj_jconf_check_addr(&jsmsg->to, jwl->aliases->dlm))		{			if((jcf = xj_jcon_get_jconf(jbc, &jsmsg->to, jwl->aliases->dlm))					!= NULL)			{				if((jsmsg->type == XJ_JOIN_JCONF) &&					!(jcf->status & XJ_JCONF_READY || 						jcf->status & XJ_JCONF_WAITING))				{					if(!xj_jcon_jconf_presence(jbc,jcf,NULL,"online"))						jcf->status = XJ_JCONF_WAITING;					else					{						// unable to join the conference 						// --- send back to SIP user a msg						xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to, 							jwl->contact_h, XJ_DMSG_ERR_JOINJCONF,							&jbc->jkey->flag);						goto step_w;					}				}				flag |= XJ_ADDRTR_CON;			}			else			{				// unable to get the conference 				// --- send back to SIP user a msg				xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to, 					jwl->contact_h, XJ_DMSG_ERR_NEWJCONF, &jbc->jkey->flag);				goto step_w;			}		}		if(jsmsg->type == XJ_JOIN_JCONF)			goto step_w;				// here will come only XJ_SEND_MESSAGE		switch(xj_jcon_is_ready(jbc,jsmsg->to.s,jsmsg->to.len,jwl->aliases->dlm))		{			case 0:				DBG("XJAB:xj_worker:%d: SENDING THE MESSAGE TO JABBER"					" NETWORK ...\n", _xj_pid);				/*** address corection ***/				sto.s = buff; 				sto.len = 0;				flag |= XJ_ADDRTR_A2B;				if(xj_address_translation(&jsmsg->to, &sto, jwl->aliases, 							flag) == 0)				{					if(xj_jcon_send_msg(jbc, sto.s, sto.len,						jsmsg->msg.s, jsmsg->msg.len,						(flag&XJ_ADDRTR_CON)?XJ_JMSG_GROUPCHAT:XJ_JMSG_CHAT)<0)													xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,							jwl->contact_h, XJ_DMSG_ERR_SENDJMSG,							&jbc->jkey->flag);				}				else					DBG("XJAB:xj_worker:%d: ERROR SENDING AS JABBER"						" MESSAGE ...\n", _xj_pid);										goto step_w;					case 1:				DBG("XJAB:xj_worker:%d:SCHEDULING THE MESSAGE."					"\n", _xj_pid);				if(xj_jcon_pool_add_jmsg(jcp, jsmsg, jbc) < 0)				{					DBG("XJAB:xj_worker:%d: SCHEDULING THE"						" MESSAGE FAILED. Message was droped.\n",_xj_pid);					xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,						jwl->contact_h, XJ_DMSG_ERR_STOREJMSG,						&jbc->jkey->flag);					goto step_w;				}				else // skip freeing the SIP message - now is in queue					goto step_y;				case 2:				xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,						jwl->contact_h, XJ_DMSG_ERR_NOREGIM,						&jbc->jkey->flag);				goto step_w;			case 3: // not joined to Jabber conference				xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to,						jwl->contact_h, XJ_DMSG_ERR_NOTJCONF,						&jbc->jkey->flag);				goto step_w;							default:				xj_send_sip_msgz(jsmsg->jkey->id, &jsmsg->to, jwl->contact_h,						XJ_DMSG_ERR_SENDJMSG, &jbc->jkey->flag);				goto step_w;		}step_v: // error connecting to Jabber server				// cleaning jab_wlist		xj_wlist_del(jwl, jsmsg->jkey, _xj_pid);		// cleaning db_query		if ((res != NULL) && (db_free_query(db_con,res) < 0))		{			DBG("XJAB:xj_worker:%d:Error while freeing"				" SQL result - worker terminated\n", _xj_pid);			return -1;		}		else			res = NULL;step_w:		if(jsmsg!=NULL)		{			xj_sipmsg_free(jsmsg);			jsmsg = NULL;		}			step_y:			 		// check for new message from ... JABBER		for(i = 0; i < jcp->len && main_loop; i++)		{			if(jcp->ojc[i] == NULL)				continue;						DBG("XJAB:xj_worker:%d: checking socket <%d>"				" ...\n", _xj_pid, jcp->ojc[i]->sock);						if(!FD_ISSET(jcp->ojc[i]->sock, &mset))				continue;			pos = nr = 0;			do			{				p = recv_buff;				if(pos != 0)				{					while(pos < nr)					{						*p = recv_buff[pos];						pos++;						p++;					}					*p = 0;					/**					 * flush out the socket - set it to nonblocking 					 */ 					flags = fcntl(jcp->ojc[i]->sock, F_GETFL, 0);					if(flags!=-1 && !(flags & O_NONBLOCK))   					{    					/* set NONBLOCK bit to enable non-blocking */    					fcntl(jcp->ojc[i]->sock, F_SETFL, flags | O_NONBLOCK);   					}				}								if((nr = read(jcp->ojc[i]->sock, p,							sizeof(recv_buff)-(p-recv_buff))) == 0					||(nr < 0 && errno != EAGAIN))				{					DBG("XJAB:xj_worker:%d: ERROR -"						" connection to jabber lost on socket <%d> ...\n",						_xj_pid, jcp->ojc[i]->sock);					xj_send_sip_msgz(jcp->ojc[i]->jkey->id, &jab_gw_name, 						jwl->contact_h, XJ_DMSG_ERR_DISCONNECTED,						&jbc->jkey->flag);					// make sure that will ckeck expired connections					ltime = jcp->ojc[i]->expire = -1;					FD_CLR(jcp->ojc[i]->sock, &set);					goto step_xx;				}				DBG("XJAB:xj_worker:%d: received: %dbytes Err:%d/EA:%d\n", 						_xj_pid, nr, errno, EAGAIN);						xj_jcon_update(jcp->ojc[i], jwl->cachet);				if(nr>0)					p[nr] = 0;				nr = strlen(recv_buff);				pos = 0;				DBG("XJAB:xj_worker: JMSG START ----------\n%.*s\n"					" JABBER: JMSGL:%d END ----------\n", nr, recv_buff, nr);						} while(xj_manage_jab(recv_buff, nr, &pos, jwl->contact_h,					jwl->aliases, jcp->ojc[i]) == 9	&& main_loop);				/**			 * flush out the socket - set it back to blocking 			 */ 			flags = fcntl(jcp->ojc[i]->sock, F_GETFL, 0);			if(flags!=-1 && (flags & O_NONBLOCK))   			{    			/* reset NONBLOCK bit to enable blocking */    			fcntl(jcp->ojc[i]->sock, F_SETFL, flags & ~O_NONBLOCK);   			}				DBG("XJAB:xj_worker:%d: msgs from socket <%d> parsed"				" ...\n", _xj_pid, jcp->ojc[i]->sock);			} // end FOR(i = 0; i < jcp->len; i++)step_x:		if(ret < 0)		{			DBG("XJAB:xj_worker:%d: SIGNAL received!!!!!!!!\n", _xj_pid);			maxfd = pipe;			FD_ZERO(&set);			FD_SET(pipe, &set);			for(i = 0; i < jcp->len; i++)			{				if(jcp->ojc[i] != NULL)				{					FD_SET(jcp->ojc[i]->sock, &set);					if( jcp->ojc[i]->sock > maxfd )						maxfd = jcp->ojc[i]->sock;				}			}		}step_xx:		if(ltime < 0 || ltime + jwl->sleept <= get_ticks())		{			ltime = get_ticks();			DBG("XJAB:xj_worker:%d: scanning for expired connection\n",				_xj_pid);			xj_worker_check_jcons(jwl, jcp, ltime, &set);		}	} // END while	DBG("XJAB:xj_worker:%d: cleaning procedure\n", _xj_pid);	return 0;} // end xj_worker_process