xjab_worker.c

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

		prc = xj_pres_list_check(jbc->plist, &ts);
		if(!prc)
			goto ready;

		if(!strncasecmp(type, "unavailable", 11))
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_manage_jab: user <%s> is offline\n", from);
#endif
			if(prc->state != XJ_PS_OFFLINE)
			{
				prc->state = XJ_PS_OFFLINE;
				goto call_pa_cbf;
			}
			goto ready;
		}
		
		if(!strncasecmp(type, "unsubscribed", 12))
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_manage_jab: user <%s> does not allow to see his"
				" presence status\n", from);
#endif
			if(prc->state != XJ_PS_REFUSED)
			{
				prc->state = XJ_PS_REFUSED;
				goto call_pa_cbf;
			}
		}
	
		// ignoring unknown types
		goto ready;
	}
/*------------------- END XMPP 'PRESENCE' HANDLING ----------------*/
	
/******************** XMPP 'IQ' HANDLING ***************************/
	if(!strncasecmp(xode_get_name(x), "iq", 2))
	{
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_manage_jab: jabber [iq] received\n");
#endif
		if(!strncasecmp(xode_get_attrib(x, "type"), "result", 6))
		{
			if((y = xode_get_tag(x, "query?xmlns=jabber:iq:roster")) == NULL)
				goto ready;
			z = xode_get_firstchild(y);
			while(z)
			{
				if(!strncasecmp(xode_get_name(z), "item", 5)
					&& (from = xode_get_attrib(z, "jid")) != NULL)
				{
					if(strchr(from, '@') == NULL)
					{ // transports
						if(!strncasecmp(from, XJ_AIM_NAME, XJ_AIM_LEN))
						{
							jbc->allowed |= XJ_NET_AIM;
#ifdef XJ_EXTRA_DEBUG
							DBG("XJAB:xj_manage_jab:AIM network available\n");
#endif
						}
						else if(!strncasecmp(from, XJ_ICQ_NAME, XJ_ICQ_LEN))
						{
							jbc->allowed |= XJ_NET_ICQ;
#ifdef XJ_EXTRA_DEBUG
							DBG("XJAB:xj_manage_jab:ICQ network available\n");
#endif
						}
						else if(!strncasecmp(from, XJ_MSN_NAME, XJ_MSN_LEN))
						{
							jbc->allowed |= XJ_NET_MSN;
#ifdef XJ_EXTRA_DEBUG
							DBG("XJAB:xj_manage_jab:MSN network available\n");
#endif
						}
						else if(!strncasecmp(from, XJ_YAH_NAME, XJ_YAH_LEN))
						{
							jbc->allowed |= XJ_NET_YAH;
#ifdef XJ_EXTRA_DEBUG
							DBG("XJAB:xj_manage_jab:YAHOO network available\n");
#endif
						}
						goto next_sibling;
					}
				}
next_sibling:
				z = xode_get_nextsibling(z);
			}
		}
		
		goto ready;
	}
/*------------------- END XMPP 'IQ' HANDLING ----------------------*/

call_pa_cbf:
	if(prc && prc->cbf)
	{
		// call the PA callback function
		tf.s = fbuf;
		tf.len = 0;
		if(xj_address_translation(&ts,&tf,als,XJ_ADDRTR_J2S)==0)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_manage_jab: calling CBF(%.*s,%d)\n",
				tf.len, tf.s, prc->state);
#endif
			(*(prc->cbf))(&tf, &tf, prc->state, prc->cbp);
		}
	}
ready:
	xode_free(x);
	return err;
}

/**
 *
 */
void xj_sig_handler(int s) 
{
	//signal(SIGTERM, xj_sig_handler);
	//signal(SIGINT, xj_sig_handler);
	//signal(SIGQUIT, xj_sig_handler);
	signal(SIGSEGV, xj_sig_handler);
	main_loop = 0;
	DBG("XJAB:xj_worker:%d: SIGNAL received=%d\n **************", _xj_pid, s);
}

/*****************************     ****************************************/

/**
 * send a SIP MESSAGE message
 * - to : destination
 * - from : origin
 * - contact : contact header
 * - msg : body of the message
 * #return : 0 on success or <0 on error
 */
int xj_send_sip_msg(str *proxy, str *to, str *from, str *msg, int *cbp)
{
	str  msg_type = { "MESSAGE", 7};
	char buf[512];
	str  tfrom;
	str  str_hdr;
	char buf1[1024];

	if( !to || !to->s || to->len <= 0 
			|| !from || !from->s || from->len <= 0 
			|| !msg || !msg->s || msg->len <= 0
			|| (cbp && *cbp!=0) )
		return -1;

	// from correction
	tfrom.len = 0;
	strncpy(buf+tfrom.len, "<sip:", 5);
	tfrom.len += 5;
	strncpy(buf+tfrom.len, from->s, from->len);
	tfrom.len += from->len;
	buf[tfrom.len++] = '>';
		
	tfrom.s = buf;
	
	// building Contact and Content-Type
	strcpy(buf1,"Content-Type: text/plain"CRLF"Contact: ");
	str_hdr.len = 24 + CRLF_LEN + 9;
	
	strncat(buf1,tfrom.s,tfrom.len);
	str_hdr.len += tfrom.len;
	
	strcat(buf1, CRLF);
	str_hdr.len += CRLF_LEN;
	str_hdr.s = buf1;
	if(cbp)
	{
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_send_sip_msg: uac callback parameter [%p==%d]\n", 
				cbp, *cbp);
#endif
		return tmb.t_request(&msg_type, 0, to, &tfrom, &str_hdr, msg, 
					     xj_tuac_callback, (void*)cbp);
	}
	else
		return tmb.t_request(&msg_type, 0, to, &tfrom, &str_hdr, msg, 0, 0);
}

/**
 * send a SIP MESSAGE message
 * - to : destination
 * - from : origin
 * - contact : contact header
 * - msg : body of the message, string terminated by zero
 * #return : 0 on success or <0 on error
 */
int xj_send_sip_msgz(str *proxy, str *to, str *from, char *msg, int *cbp)
{
	str tstr;
	int n;

	if(!to || !from || !msg || (cbp && *cbp!=0))
		return -1;

	tstr.s = msg;
	tstr.len = strlen(msg);
	if((n = xj_send_sip_msg(proxy, to, from, &tstr, cbp)) < 0)
		DBG("XJAB: jab_send_sip_msgz: ERROR SIP MESSAGE wasn't sent to"
			" [%.*s]...\n", to->len, to->s);
#ifdef XJ_EXTRA_DEBUG
	else
		DBG("XJAB: jab_send_sip_msgz: SIP MESSAGE was sent to [%.*s]...\n",
			to->len, to->s);
#endif
	return n;
}

/**
 * send disconnected info to all SIP users associated with worker idx
 * and clean the entries from wlist
 */
int xj_wlist_clean_jobs(xj_wlist jwl, int idx, int fl)
{
	xj_jkey p;
	if(jwl==NULL || idx < 0 || idx >= jwl->len || !jwl->workers[idx].sip_ids)
		return -1;
	lock_set_get(jwl->sems, idx);
	while((p=(xj_jkey)delpos234(jwl->workers[idx].sip_ids, 0))!=NULL)
	{
		if(fl)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_wlist_send_info: sending disconnect message"
				" to <%.*s>\n",	p->id->len, p->id->s);
#endif
			xj_send_sip_msgz(_PADDR(jwl), p->id, &jab_gw_name,
					XJ_DMSG_INF_DISCONNECTED, NULL);
		}
		jwl->workers[idx].nr--;
		xj_jkey_free_p(p);
	}
	lock_set_release(jwl->sems, idx);
	return 0;
}


/**
 * callback function for TM
 */
void xj_tuac_callback( struct cell *t, int type, struct tmcb_params *ps)
{
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB: xj_tuac_callback: completed with status %d\n", ps->code);
#endif
	if(!ps->param)
	{
		DBG("XJAB: m_tuac_callback: parameter not received\n");
		return;
	}
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB: xj_tuac_callback: parameter [%p : ex-value=%d]\n", ps->param,
					*((int*)ps->param) );
#endif
	if(ps->code < 200 || ps->code >= 300)
	{
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB: xj_tuac_callback: no 2XX return code - connection set"
			" as expired \n");
#endif
		*((int*)ps->param) = XJ_FLAG_CLOSE;
	}
}

/**
 * check for expired connections
 */
void xj_worker_check_jcons(xj_wlist jwl, xj_jcon_pool jcp, int ltime, fd_set *pset)
{
	int i;
	xj_jconf jcf;
	
	for(i = 0; i < jcp->len && main_loop; i++)
	{
		if(jcp->ojc[i] == NULL)
			continue;
		if(jcp->ojc[i]->jkey->flag==XJ_FLAG_OPEN &&
			jcp->ojc[i]->expire > ltime)
			continue;
			
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_worker:%d: connection expired for <%.*s> \n",
			_xj_pid, jcp->ojc[i]->jkey->id->len, jcp->ojc[i]->jkey->id->s);
#endif
		xj_send_sip_msgz(_PADDR(jwl), jcp->ojc[i]->jkey->id, &jab_gw_name,
				XJ_DMSG_INF_JOFFLINE, NULL);
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_worker:%d: connection's close flag =%d\n",
			_xj_pid, jcp->ojc[i]->jkey->flag);
#endif
		// CLEAN JAB_WLIST
		xj_wlist_del(jwl, jcp->ojc[i]->jkey, _xj_pid);

		// looking for open conference rooms
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_worker:%d: having %d open conferences\n", 
				_xj_pid, jcp->ojc[i]->nrjconf);
#endif
		while(jcp->ojc[i]->nrjconf > 0)
		{
			if((jcf=delpos234(jcp->ojc[i]->jconf,0))!=NULL)
			{
				// get out of room
				xj_jcon_jconf_presence(jcp->ojc[i],jcf, "unavailable", NULL);
				xj_jconf_free(jcf);
			}
			jcp->ojc[i]->nrjconf--;
		}

		// send offline presence to all subscribers
		if(jcp->ojc[i]->plist)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker:%d: sending 'terminated' status to SIP"
					" subscriber\n", _xj_pid);
#endif
			xj_pres_list_notifyall(jcp->ojc[i]->plist,
					XJ_PS_TERMINATED);
		}
		FD_CLR(jcp->ojc[i]->sock, pset);
		xj_jcon_disconnect(jcp->ojc[i]);
		xj_jcon_free(jcp->ojc[i]);
		jcp->ojc[i] = NULL;
	}
}

/**
 * check if there are msg to send or delete from queue
 */
void xj_worker_check_qmsg(xj_wlist jwl, xj_jcon_pool jcp)
{
	int i, flag;
	str sto;
	char buff[1024];

	if(!jwl || !jcp)
		return;

	/** check the msg queue AND if the target connection 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())
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker_check_qmsg:%d: message to %.*s is expired\n",
				_xj_pid, jcp->jmqueue.jsm[i]->to.len, 
				jcp->jmqueue.jsm[i]->to.s);
#endif
			xj_send_sip_msgz(_PADDR(jwl), jcp->jmqueue.jsm[i]->jkey->id, 
					&jcp->jmqueue.jsm[i]->to, 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;
		}
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_worker_check_qmsg:%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]);
#endif
		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 correction ***/
		flag = XJ_ADDRTR_S2J;
		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 */
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker_check_qmsg:%d: SENDING the message from"
				" local queue to Jabber network ...\n", _xj_pid);
#endif
			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_check_qmsg:%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);
	}
}


/**
 * update or register a presence watcher
 */
void xj_worker_check_watcher(xj_wlist jwl, xj_jcon_pool jcp,
				xj_jcon jbc, xj_sipmsg jsmsg)
{
	str sto;
	char buff[1024];
	xj_pres_cell prc = NULL;

	if(!jwl || !jcp || !jbc || !jsmsg)
		return;

	if(!jsmsg->cbf)
	{
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_worker_check_watcher:%d: NULL PA callback"
			" function\n", _xj_pid);
#endif
		return;
	}

	if(!xj_jconf_check_addr(&jsmsg->to, jwl->aliases->dlm))
	{ // is for a conference - ignore?!?!
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:xj_worker_check_watcher:%d: presence request for a"
			" conference.\n", _xj_pid);
#endif
		// set as offline
		(*(jsmsg->cbf))(&jsmsg->to, &jsmsg->to, XJ_PS_OFFLINE, jsmsg->p);
		return;
	}
			
	sto.s = buff; 
	sto.len = 0;

	if(xj_address_translation(&jsmsg->to, &sto, jwl->aliases, 
		XJ_ADDRTR_S2J) == 0)
	{
		prc = xj_pres_list_check(jbc->plist, &sto);
		if(!prc)
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker_check_watcher:%d: NEW presence"
				" cell for %.*s.\n", _xj_pid, sto.len, sto.s);
#endif
			prc = xj_pres_cell_new();
			if(!prc)
			{
				DBG("XJAB:xj_worker_check_watcher:%d: cannot create a presence"
					" cell for %.*s.\n", _xj_pid, sto.len, sto.s);
				return;
			}
			if(xj_pres_cell_init(prc, &sto, jsmsg->cbf, jsmsg->p)<0)
			{
				DBG("XJAB:xj_worker_check_watcher:%d: cannot init the presence"
					" cell for %.*s.\n", _xj_pid, sto.len, sto.s);
				xj_pres_cell_free(prc);
				return;
			}
			if((prc = xj_pres_list_add(jbc->plist, prc))==NULL)
			{
				DBG("XJAB:xj_worker_check_watcher:%d: cannot add the presence"
					" cell for %.*s.\n", _xj_pid, sto.len, sto.s);
				return;
			}
			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);
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xj_worker_check_watcher:%d: calling CBF(%.*s,%d)\n",
				_xj_pid, jsmsg->to.len, jsmsg->to.s, prc->state);
#endif
			// send presence info to SIP subscriber
			(*(prc->cbf))(&jsmsg->to, &jsmsg->to, prc->state, prc->cbp);
		}
	}
}

/*****************************     ****************************************/