jabber.c

性能优秀的SIP Proxy

			}
		break;
		default:
			DBG("XJAB:xjab_manage_sipmsg: ERROR:strange SIP msg type!\n");
			goto error;
	}

	// if is for going ONLINE/OFFLINE we do not need the destination
	if(type==XJ_GO_ONLINE || type==XJ_GO_OFFLINE)
		goto prepare_job;
	
	// determination of destination
	// - try to get it from new_uri, r-uri or to hdr, but check it against
	// jdomain and aliases
	dst.len = 0;
	if( msg->new_uri.len > 0)
	{
		dst.s = msg->new_uri.s;
		dst.len = msg->new_uri.len;
		if(xj_wlist_check_aliases(jwl, &dst))
			dst.len = 0;
#ifdef XJ_EXTRA_DEBUG
		else
			DBG("XJAB:xjab_manage_sipmsg: using NEW URI for destination\n");
#endif
	}
	
	if (dst.len == 0 &&  msg->first_line.u.request.uri.s != NULL
			&& msg->first_line.u.request.uri.len > 0 )
	{
		dst.s = msg->first_line.u.request.uri.s;
		dst.len = msg->first_line.u.request.uri.len;
		if(xj_wlist_check_aliases(jwl, &dst))
			dst.len = 0;
#ifdef XJ_EXTRA_DEBUG
		else
			DBG("XJAB:xjab_manage_sipmsg: using R-URI for destination\n");
#endif
	}

	if(dst.len == 0 && msg->to->parsed)
	{
		dst.s = ((struct to_body*)msg->to->parsed)->uri.s;
		dst.len = ((struct to_body*)msg->to->parsed)->uri.len;
		if(dst.s == NULL || xj_wlist_check_aliases(jwl, &dst))
			dst.len = 0;
#ifdef XJ_EXTRA_DEBUG
		else
			DBG("XJAB:xjab_manage_sipmsg: using TO-URI for destination\n");
#endif
	}
	
	if(dst.len == 0)
	{
		DBG("XJAB:xjab_manage_sipmsg: destination not found in SIP message\n");
		goto error;
	}
	
	/** skip 'sip:' and parameters in destination address */
	if(xj_extract_aor(&dst, 1))
	{
		DBG("ERROR:xjab_manage_sipmsg: cannot get AoR for destination\n");
		goto error;
	}
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xjab_manage_sipmsg: DESTINATION after correction [%.*s].\n",
				dst.len, dst.s);
#endif
	
prepare_job:
	//putting the SIP message parts in share memory to be accessible by workers
    jsmsg = (xj_sipmsg)shm_malloc(sizeof(t_xj_sipmsg));
	memset(jsmsg, 0, sizeof(t_xj_sipmsg));
    if(jsmsg == NULL)
    	return -1;
	
	switch(type)
	{
		case XJ_SEND_MESSAGE:
			jsmsg->msg.len = body.len;
			if((jsmsg->msg.s = (char*)shm_malloc(jsmsg->msg.len+1)) == NULL)
			{
				shm_free(jsmsg);
				goto error;
			}
			strncpy(jsmsg->msg.s, body.s, jsmsg->msg.len);
		break;
		case XJ_GO_ONLINE:
		case XJ_GO_OFFLINE:
			dst.len = 0;
			dst.s = 0;
		case XJ_JOIN_JCONF:
		case XJ_EXIT_JCONF:
			jsmsg->msg.len = 0;
			jsmsg->msg.s = NULL;
		break;
		default:
			DBG("XJAB:xjab_manage_sipmsg: this SHOULD NOT appear\n");
			shm_free(jsmsg);
			goto error;
	}
	if(dst.len>0)
	{
		jsmsg->to.len = dst.len;
		if((jsmsg->to.s = (char*)shm_malloc(jsmsg->to.len+1))==NULL)
		{
			if(type == XJ_SEND_MESSAGE)
				shm_free(jsmsg->msg.s);
			shm_free(jsmsg);
			goto error;
		}
		strncpy(jsmsg->to.s, dst.s, jsmsg->to.len);
	}
	else
	{
		jsmsg->to.len = 0;
		jsmsg->to.s   = 0;
	}

	jsmsg->jkey = p;
	jsmsg->type = type;
	//jsmsg->jkey->hash = jkey.hash;

	DBG("XJAB:xjab_manage_sipmsg:%d: sending <%p> to worker through <%d>\n",
			getpid(), jsmsg, pipe);
	// sending the SHM pointer of SIP message to the worker
	fl = write(pipe, &jsmsg, sizeof(jsmsg));
	if(fl != sizeof(jsmsg))
	{
		DBG("XJAB:xjab_manage_sipmsg: error when writing to worker pipe!\n");
		if(type == XJ_SEND_MESSAGE)
			shm_free(jsmsg->msg.s);
		shm_free(jsmsg->to.s);
		shm_free(jsmsg);
		goto error;
	}
	
	return 1;
error:
	return -1;
}

/**
 * destroy function of module
 */
void destroy(void)
{
	int i;
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB: Unloading module ...\n");
#endif
	if(pipes)
	{ // close the pipes
		for(i = 0; i < nrw; i++)
		{
			if(pipes[i])
			{
				close(pipes[i][0]);
				close(pipes[i][1]);
			}
			pkg_free(pipes[i]);
		}
		pkg_free(pipes);
	}
	// cleaning MySQL connections
	if(db_con != NULL)
	{
		for(i = 0; i<nrw; i++)
			jabber_dbf.close(db_con[i]);
		shm_free(db_con);
	}
			
	xj_wlist_free(jwl);
	DBG("XJAB: Unloaded ...\n");
}

/**
 * register a watcher function for a Jabber user' presence
 */
void xj_register_watcher(str *from, str *to, void *cbf, void *pp)
{
	xj_sipmsg jsmsg = NULL;
	t_xj_jkey jkey, *jp;
	int pipe, fl;
	str from_uri, to_uri;

	if(!to || !from || !cbf)
		return;

#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_register_watcher: from=[%.*s] to=[%.*s]\n", from->len,
	    from->s, to->len, to->s);
#endif
	from_uri.s = from->s;
	from_uri.len = from->len;
	if(xj_extract_aor(&from_uri, 0))
	{
		DBG("ERROR:xjab_manage_sipmsg: cannot get AoR from FROM header\n");
		goto error;
	}

	jkey.hash = xj_get_hash(&from_uri, NULL);
	jkey.id = &from_uri;

	if((pipe = xj_wlist_get(jwl, &jkey, &jp)) < 0)
	{
		DBG("XJAB:xj_register_watcher: cannot find pipe of the worker!\n");
		goto error;
	}
	
	//putting the SIP message parts in share memory to be accessible by workers
	jsmsg = (xj_sipmsg)shm_malloc(sizeof(t_xj_sipmsg));
	memset(jsmsg, 0, sizeof(t_xj_sipmsg));
	if(jsmsg == NULL)
		goto error;
	
	jsmsg->msg.len = 0;
	jsmsg->msg.s = NULL;
	
	to_uri.s = to->s;
	to_uri.len = to->len;
	/** skip 'sip:' and parameters in destination address */
	if(xj_extract_aor(&to_uri, 1))
	{
		DBG("ERROR:xjab_manage_sipmsg: cannot get AoR for destination\n");
		goto error;
	}
#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_register_watcher: DESTINATION after correction [%.*s].\n",
	    to_uri.len, to_uri.s);
#endif

	jsmsg->to.len = to_uri.len;
	if((jsmsg->to.s = (char*)shm_malloc(jsmsg->to.len+1)) == NULL)
	{
		if(jsmsg->msg.s)
			shm_free(jsmsg->msg.s);
		shm_free(jsmsg);
		goto error;
	}
	strncpy(jsmsg->to.s, to_uri.s, jsmsg->to.len);
	jsmsg->to.s[jsmsg->to.len] = '\0';

	jsmsg->jkey = jp;
	jsmsg->type = XJ_REG_WATCHER;
	//jsmsg->jkey->hash = jkey.hash;
	
	jsmsg->cbf = (pa_callback_f)cbf;
	jsmsg->p = pp;

#ifdef XJ_EXTRA_DEBUG
	DBG("XJAB:xj_register_watcher:%d: sending <%p> to worker through <%d>\n",
	    getpid(), jsmsg, pipe);
#endif
	// sending the SHM pointer of SIP message to the worker
	fl = write(pipe, &jsmsg, sizeof(jsmsg));
	if(fl != sizeof(jsmsg))
	{
		DBG("XJAB:xj_register_watcher: error when writing to worker pipe!\n");
		if(jsmsg->msg.s)
			shm_free(jsmsg->msg.s);
		shm_free(jsmsg->to.s);
		shm_free(jsmsg);
		goto error;
	}
	
 error:
	return;
}

/**
 * unregister a watcher for a Jabber user' presence
 */
void xj_unregister_watcher(str *from, str *to, void *cbf, void *pp)
{
	if(!to || !from)
		return;
}

/**
 * check if all SER2Jab workers are still alive
 * - if not, try to launch new ones
 */
void xjab_check_workers(int mpid)
{
	int i, n, stat;
	//DBG("XJAB:%d:xjab_check_workers: time=%d\n", mpid, get_ticks());
	if(!jwl || jwl->len <= 0)
		return;
	for(i=0; i < jwl->len; i++)
	{
		if(jwl->workers[i].pid > 0)
		{
			stat = 0;
			n = waitpid(jwl->workers[i].pid, &stat, WNOHANG);
			if(n == 0 || n!=jwl->workers[i].pid)
				continue;
		
			LOG(L_ERR,"XJAB:xjab_check_workers: worker[%d][pid=%d] has exited"
				" - status=%d err=%d errno=%d\n", i, jwl->workers[i].pid, 
				stat, n, errno);
			xj_wlist_clean_jobs(jwl, i, 1);
			xj_wlist_set_pid(jwl, -1, i);
		}
		
#ifdef XJ_EXTRA_DEBUG
		DBG("XJAB:%d:xjab_check_workers: create a new worker[%d]\n", mpid, i);
#endif
		if ( (stat=fork())<0 )
		{
#ifdef XJ_EXTRA_DEBUG
			DBG("XJAB:xjab_check_workers: error - cannot launch new"
				" worker[%d]\n", i);
#endif
			LOG(L_ERR, "XJAB:xjab_check_workers: error - worker[%d] lost"
				" forever \n", i);
			return;
		}
		if (stat == 0)
		{
			if(xj_wlist_set_pid(jwl, getpid(), i) < 0)
			{
				LOG(L_ERR, "XJAB:xjab_check_workers: error setting new"
					" worker's pid - w[%d]\n", i);
				return;
			}
			xj_worker_process(jwl,jaddress,jport,i,db_con[i], &jabber_dbf);
			exit(0);
		}
	}			
}

#ifdef HAVE_IHTTP
/**
 * Module's information retrieval - function to use with iHttp module
 *
 */ 
int xjab_mod_info(ih_req_p _irp, void *_p, char *_bb, int *_bl, 
		char *_hb, int *_hl)
{
	if(!_irp || !_bb || !_bl || *_bl <= 0 || !_hb || !_hl || *_hl <= 0)
		return -1;
	*_hl = 0;
	*_hb = 0;
	
	strcpy(_bb, "<h4>SER2Jabber Gateway</h4>");
	strcat(_bb, "<br>Module parameters:<br>");
	strcat(_bb, "<br> -- db table = ");
	strcat(_bb, db_table);
	strcat(_bb, "<br> -- workers = ");
	strcat(_bb, int2str(nrw, NULL));
	strcat(_bb, "<br> -- max jobs per worker = ");
	strcat(_bb, int2str(max_jobs, NULL));

	strcat(_bb, "<br> -- jabber server address = ");
	strcat(_bb, jaddress);
	strcat(_bb, "<br> -- jabber server port = ");
	strcat(_bb, int2str(jport, NULL));

	strcat(_bb, "<br> -- aliases = ");
	strcat(_bb, (jaliases)?jaliases:"NULL");
	strcat(_bb, "<br> -- jabber domain = ");
	strcat(_bb, (jdomain)?jdomain:"NULL");
	strcat(_bb, "<br> -- proxy address = ");
	strcat(_bb, (proxy)?proxy:"NULL");

	strcat(_bb, "<br> -- delay time = ");
	strcat(_bb, int2str(delay_time, NULL));
	strcat(_bb, "<br> -- sleep time = ");
	strcat(_bb, int2str(sleep_time, NULL));
	strcat(_bb, "<br> -- cache time = ");
	strcat(_bb, int2str(cache_time, NULL));
	strcat(_bb, "<br> -- check time = ");
	strcat(_bb, int2str(check_time, NULL));
	
	*_bl = strlen(_bb);

	return 0;
}

/**
 * SER2Jab connection management - function to use with iHttp module
 * - be aware of who is able to use the ihttp because he can close any 
 *   open connection between SER and Jabber server
 */ 
int xjab_connections(ih_req_p _irp, void *_p, char *_bb, int *_bl, 
		char *_hb, int *_hl)
{
	t_xj_jkey jkey, *p;
	str _u;
	ih_param_p _ipp = NULL;
	int idx, i, maxcount;
	char *cp;

	if(!_irp || !_bb || !_bl || *_bl <= 0 || !_hb || !_hl || *_hl <= 0)
		return -1;
	
	*_hl = 0;
	*_hb = 0;
	idx = -1;
	strcpy(_bb, "<h4>Active XMPP connections</h4>");
	
	if(_irp->params)
	{
		strcat(_bb, "<br><b>Close action is alpha release!</b><br>");
		_ipp = _irp->params;
		i = 0;
		while(_ipp)
		{
			switch(_ipp->name[0])
			{
				case 'w':
					idx = 0;
					cp = _ipp->value;
					while(*cp && *cp>='0' && *cp<='9')
					{
						idx = idx*10 + *cp-'0';
						cp++;
					}
					i++;
				break;
				case 'u':
					_u.s = _ipp->value;
					_u.len = strlen(_ipp->value);
					jkey.id = &_u;
					i++;
				break;
				case 'i':
					jkey.hash = 0;
					cp = _ipp->value;
					while(*cp && *cp>='0' && *cp<='9')
					{
						jkey.hash = jkey.hash*10 + *cp-'0';
						cp++;
					}
					i++;
				break;
				
			}
			_ipp = _ipp->next;
		}
		if(i!=3 || idx < 0 || idx >= jwl->len)
		{
			strcat(_bb, "<br><b><i>Bad parameters!</i></b>\n");
		}
		else
		{
			strcat(_bb, "<br><b><i>The connection of [");
			strcat(_bb, _u.s);

			if(xj_wlist_set_flag(jwl, &jkey, XJ_FLAG_CLOSE) < 0)
				strcat(_bb, "] does not exist!</i></b>\n");
			else
				strcat(_bb, "] was scheduled for closing!</i></b>\n");
		}
		*_bl = strlen(_bb);

		return 0;
	}
	
	if(jwl!=NULL && jwl->len > 0 && jwl->workers!=NULL)
	{
		for(idx=0; idx<jwl->len; idx++)
		{
			strcat(_bb, "<br><b><i>Worker[");
			strcat(_bb, int2str(idx, NULL));
			strcat(_bb, "]</i></b> &nbsp;&nbsp;pid=");
			strcat(_bb, int2str(jwl->workers[idx].pid, NULL));
			strcat(_bb, " &nbsp;&nbsp;nr of jobs=");
			strcat(_bb, int2str(jwl->workers[idx].nr, NULL));
			if(!jwl->workers[idx].sip_ids)
				continue;
			lock_set_get(jwl->sems, idx);
			maxcount = count234(jwl->workers[idx].sip_ids);
			for (i = 0; i < maxcount; i++) 
			{
				p = (xj_jkey)index234(jwl->workers[idx].sip_ids, i);
				if(p == NULL)
					continue;
				strcat(_bb, "<br>&nbsp;&nbsp;&nbsp;");
				strcat(_bb, int2str(i, NULL));
				strcat(_bb, ".&nbsp;&nbsp;&nbsp;");
				strcat(_bb, "<a href=\"xjabc?w=");
				strcat(_bb, int2str(idx, NULL));
				strcat(_bb, "&i=");
				strcat(_bb, int2str(p->hash, NULL));
				strcat(_bb, "&u=");
				strncat(_bb, p->id->s, p->id->len);
				strcat(_bb, "\">close</a>");
				strcat(_bb, "&nbsp;&nbsp;&nbsp;");
				strcat(_bb, int2str(p->hash, NULL));
				strcat(_bb, "&nbsp;&nbsp;&nbsp;");
				strncat(_bb, p->id->s, p->id->len);
			}
			lock_set_release(jwl->sems, idx);
		}
	}
	
	*_bl = strlen(_bb);

	return 0;
}

#endif // HAVE_IHTTP