xjab_worker.c

SIP Express Router, Linux下的SIP代理服务器,小巧实用,开发测试VoIP设备和应用的必备.

		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(_PADDR(jwl),jsmsg->jkey->id,&jsmsg->to,
							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(_PADDR(jwl), jsmsg->jkey->id, &jsmsg->to,
						XJ_DMSG_ERR_NEWJCONF, &jbc->jkey->flag);
				goto step_w;
			}
		}
		if(jsmsg->type != XJ_SEND_MESSAGE)
			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:
#ifdef XJ_EXTRA_DEBUG
				DBG("XJAB:xj_worker:%d: SENDING the message to Jabber"
					" network ...\n", _xj_pid);
#endif
				/*** address correction ***/
				sto.s = buff; 
				sto.len = 0;
				flag |= XJ_ADDRTR_S2J;
				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(_PADDR(jwl),jsmsg->jkey->id,&jsmsg->to,
							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:
#ifdef XJ_EXTRA_DEBUG
				DBG("XJAB:xj_worker:%d:SCHEDULING the message.\n", _xj_pid);
#endif
				if(xj_jcon_pool_add_jmsg(jcp, jsmsg, jbc) < 0)
				{
					DBG("XJAB:xj_worker:%d: SCHEDULING the message FAILED."
							" Message was dropped.\n",_xj_pid);
					xj_send_sip_msgz(_PADDR(jwl), jsmsg->jkey->id, &jsmsg->to,
						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(_PADDR(jwl), jsmsg->jkey->id, &jsmsg->to,
						XJ_DMSG_ERR_NOREGIM, &jbc->jkey->flag);
				goto step_w;
			case 3: // not joined to Jabber conference
				xj_send_sip_msgz(_PADDR(jwl), jsmsg->jkey->id, &jsmsg->to,
						XJ_DMSG_ERR_NOTJCONF, &jbc->jkey->flag);
				goto step_w;
				
			default:
				xj_send_sip_msgz(_PADDR(jwl), jsmsg->jkey->id, &jsmsg->to,
						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) && (dbf->free_result(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;
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker:%d: checking socket <%d>"
				" ...\n", _xj_pid, jcp->ojc[i]->sock);
#endif
			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(_PADDR(jwl), jcp->ojc[i]->jkey->id,
						&jab_gw_name,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;
				}
#ifdef XJ_EXTRA_DEBUG
				DBG("XJAB:xj_worker:%d: received: %dbytes Err:%d/EA:%d\n", 
						_xj_pid, nr, errno, EAGAIN);
#endif
				xj_jcon_update(jcp->ojc[i], jwl->cachet);

				if(nr>0)
					p[nr] = 0;
				nr = strlen(recv_buff);
				pos = 0;
#ifdef XJ_EXTRA_DEBUG
				DBG("XJAB:xj_worker: JMSG START ----------\n%.*s\n"
					" JABBER: JMSGL:%d END ----------\n", nr, recv_buff, nr);
#endif
			} while(xj_manage_jab(recv_buff, nr, &pos, 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);
   			}
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker:%d: msgs from socket <%d> parsed"
				" ...\n", _xj_pid, jcp->ojc[i]->sock);	
#endif
		} // 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();
#ifdef XJ_EXTRA_DEBUG
			//DBG("XJAB:xj_worker:%d: scanning for expired connection\n",
			//	_xj_pid);
#endif
			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


/**
 * parse incoming message from Jabber server
 */
int xj_manage_jab(char *buf, int len, int *pos, xj_jalias als, xj_jcon jbc)
{
	int j, err=0;
	char *p, *to, *from, *msg, *type, *emsg, *ecode, lbuf[4096], fbuf[128];
	xj_jconf jcf = NULL;
	str ts, tf;
	xode x, y, z;
	str *sid;
	xj_pres_cell prc = NULL;

	if(!jbc)
		return -1;

	sid = jbc->jkey->id;	
	x = xode_from_strx(buf, len, &err, &j);
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_parse_jab: XODE ret:%d pos:%d\n", err, j);
#endif	
	if(err && pos != NULL)
		*pos= j;
	if(x == NULL)
		return -1;
	lbuf[0] = 0;
	ecode = NULL;

/******************** XMPP 'MESSAGE' HANDLING **********************/
	
	if(!strncasecmp(xode_get_name(x), "message", 7))
	{
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_manage_jab: jabber [message] received\n");
#endif
		if((to = xode_get_attrib(x, "to")) == NULL)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_manage_jab: missing 'to' attribute\n");
#endif
			err = -1;
			goto ready;
		}
		if((from = xode_get_attrib(x, "from")) == NULL)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_manage_jab: missing 'from' attribute\n");
#endif
			err = -1;
			goto ready;
		}
		if((y = xode_get_tag(x, "body")) == NULL
				|| (msg = xode_get_data(y)) == NULL)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_manage_jab: missing 'body' of message\n");
#endif
			err = -1;
			goto ready;
		}
		type = xode_get_attrib(x, "type");
		if(type != NULL && !strncasecmp(type, "error", 5))
		{
			if((y = xode_get_tag(x, "error")) == NULL
					|| (emsg = xode_get_data(y)) == NULL)
				strcpy(lbuf, "{Error sending following message} - ");
			else
			{
				ecode = xode_get_attrib(y, "code");
				strcpy(lbuf, "{Error (");
				if(ecode != NULL)
				{
					strcat(lbuf, ecode);
					strcat(lbuf, " - ");
				}
				strcat(lbuf, emsg);
				strcat(lbuf, ") when trying to send following message}");
			}

		}

		// is from a conference?!?!
		if((jcf=xj_jcon_check_jconf(jbc, from))!=NULL)
		{
			if(lbuf[0] == 0)
			{
				p = from + strlen(from);
				while(p>from && *p != '/')
					p--;
				if(*p == '/')
				{
					if(jcf->nick.len>0 
						&& strlen(p+1) == jcf->nick.len
						&& !strncasecmp(p+1, jcf->nick.s, jcf->nick.len))
					{
#ifdef XJ_EXTRA_DEBUG
						DBG("XJAB:xj_manage_jab: message sent by myself\n");
#endif
						goto ready;
					}
					lbuf[0] = '[';
					lbuf[1] = 0;
					strcat(lbuf, p+1);
					strcat(lbuf, "] ");
				}
			}
			else
			{
				jcf->status = XJ_JCONF_NULL;
				xj_jcon_jconf_presence(jbc,jcf,NULL,"online");
			}
			strcat(lbuf, msg);
			ts.s = lbuf;
			ts.len = strlen(lbuf);
	
			if(xj_send_sip_msg(als->proxy, sid, &jcf->uri, &ts,
						&jbc->jkey->flag)<0)
				DBG("XJAB:xj_manage_jab: ERROR SIP MESSAGE was not sent!\n");
#ifdef XJ_EXTRA_DEBUG
			else
				DBG("XJAB:xj_manage_jab: SIP MESSAGE was sent!\n");
#endif
			goto ready;
		}

		strcat(lbuf, msg);
		ts.s = from;
		ts.len = strlen(from);
		tf.s = fbuf;
		tf.len = 0;
		if(xj_address_translation(&ts, &tf, als, XJ_ADDRTR_J2S) == 0)
		{
			ts.s = lbuf;
			ts.len = strlen(lbuf);
	
			if(xj_send_sip_msg(als->proxy, sid, &tf, &ts, &jbc->jkey->flag)<0)
				DBG("XJAB:xj_manage_jab: ERROR SIP MESSAGE was not sent ...\n");
#ifdef XJ_EXTRA_DEBUG
			else
				DBG("XJAB:xj_manage_jab: SIP MESSAGE was sent.\n");
#endif
		}
		goto ready;
	}
/*------------------- END 'MESSAGE' HANDLING ----------------------*/
	
/******************** XMPP 'PRESENCE' HANDLING *********************/
	if(!strncasecmp(xode_get_name(x), "presence", 8))
	{
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_manage_jab: jabber [presence] received\n");
#endif
		type = xode_get_attrib(x, "type");
		from = xode_get_attrib(x, "from");
		if(from == NULL)
			goto ready;
		ts.s = from;
		p = from;
		while(p<from + strlen(from) && *p != '/')
					p++;
		if(*p == '/')
			ts.len = p - from;
		else
			ts.len = strlen(from);

		if(type == NULL || !strncasecmp(type, "online", 6)
			|| !strncasecmp(type, "available", 9))
		{
			if(strchr(from, '@') == NULL)
			{
				if(!strncasecmp(from, XJ_AIM_NAME, XJ_AIM_LEN))
				{
					jbc->ready |= XJ_NET_AIM;
#ifdef XJ_EXTRA_DEBUG
					DBG("XJAB:xj_manage_jab: AIM network ready\n");
#endif
				}
				else if(!strncasecmp(from, XJ_ICQ_NAME, XJ_ICQ_LEN))
				{
					jbc->ready |= XJ_NET_ICQ;
#ifdef XJ_EXTRA_DEBUG
					DBG("XJAB:xj_manage_jab: ICQ network ready\n");
#endif
				}
				else if(!strncasecmp(from, XJ_MSN_NAME, XJ_MSN_LEN))
				{
					jbc->ready |= XJ_NET_MSN;
#ifdef XJ_EXTRA_DEBUG
					DBG("XJAB:xj_manage_jab: MSN network ready\n");
#endif
				}
				else if(!strncasecmp(from, XJ_YAH_NAME, XJ_YAH_LEN))
				{
					jbc->ready |= XJ_NET_YAH;
#ifdef XJ_EXTRA_DEBUG
					DBG("XJAB:xj_manage_jab: YAHOO network ready\n");
#endif
				}
			}
			else if((jcf=xj_jcon_check_jconf(jbc, from))!=NULL)
			{
				jcf->status = XJ_JCONF_READY;
#ifdef XJ_EXTRA_DEBUG
				DBG("XJAB:xj_manage_jab: %s conference ready\n", from);
#endif
			}
			else
			{
#ifdef XJ_EXTRA_DEBUG
				DBG("XJAB:xj_manage_jab: user <%.*s> is online\n",ts.len,ts.s);
#endif
				prc = xj_pres_list_check(jbc->plist, &ts);
				if(prc)
				{
					if(prc->state != XJ_PS_ONLINE)
					{
						prc->state = XJ_PS_ONLINE;
						goto call_pa_cbf;
					}
				}
				else
				{
#ifdef XJ_EXTRA_DEBUG
					DBG("XJAB:xj_manage_jab: user state received - creating"
						" presence cell for [%.*s]\n", ts.len, ts.s);
#endif
					prc = xj_pres_cell_new();
					if(prc == NULL)
					{
						DBG("XJAB:xj_manage_jab: cannot create presence"
							" cell for [%s]\n", from);
						goto ready;
					}
					if(xj_pres_cell_init(prc, &ts, NULL, NULL)<0)
					{
						DBG("XJAB:xj_manage_jab: cannot init presence"
							" cell for [%s]\n", from);
						xj_pres_cell_free(prc);
						goto ready;
					}
					prc = xj_pres_list_add(jbc->plist, prc);
					if(prc)
					{
						prc->state = XJ_PS_ONLINE;
						goto call_pa_cbf;
					}
				}
			}
			goto ready;
		}
		
		if(strchr(from, '@') == NULL)
			goto ready;
	
		
		if(!strncasecmp(type, "error", 5))
		{
			if((jcf=xj_jcon_check_jconf(jbc, from))!=NULL)
			{
				tf.s = from;
				tf.len = strlen(from);
				if((y = xode_get_tag(x, "error")) == NULL)
					goto ready;
				if ((p = xode_get_attrib(y, "code")) != NULL
						&& atoi(p) == 409)
				{
					xj_send_sip_msgz(als->proxy, sid, &tf,
							XJ_DMSG_ERR_JCONFNICK, &jbc->jkey->flag);
					goto ready;
				}
				xj_send_sip_msgz(als->proxy,sid,&tf,XJ_DMSG_ERR_JCONFREFUSED,
						&jbc->jkey->flag);
			}
			goto ready;
		}
		if(type!=NULL && !strncasecmp(type, "subscribe", 9))
		{
			xj_jcon_send_presence(jbc, from, "subscribed", NULL, NULL);
			goto ready;
		}