/* use tree tn, match var, by mode, output in avp params */
static int mt_match(struct sip_msg *msg, gparam_t *tn, gparam_t *var,
		gparam_t *mode)
{
	str tname;
	str tomatch;
	int mval;
	m_tree_t *tr = NULL;
	
	if(msg==NULL)
	{
		LM_ERR("received null msg\n");
		return -1;
	}

	if(fixup_get_svalue(msg, tn, &tname)<0)
	{
		LM_ERR("cannot get the tree name\n");
		return -1;
	}
	if(fixup_get_svalue(msg, var, &tomatch)<0)
	{
		LM_ERR("cannot get the match var\n");
		return -1;
	}
	if(fixup_get_ivalue(msg, mode, &mval)<0)
	{
		LM_ERR("cannot get the mode\n");
		return -1;
	}
	
again:
	lock_get( mt_lock );
	if (mt_reload_flag) {
		lock_release( mt_lock );
		sleep_us(5);
		goto again;
	}
	mt_tree_refcnt++;
	lock_release( mt_lock );

	tr = mt_get_tree(&tname);
	if(tr==NULL)
	{
		/* no tree with such name*/
		goto error;
	}

	if(mt_match_prefix(msg, tr, &tomatch, mval)<0)
	{
		LM_INFO("no prefix found in [%.*s] for [%.*s]\n",
				tname.len, tname.s,
				tomatch.len, tomatch.s);
		goto error;
	}
	
	lock_get( mt_lock );
	mt_tree_refcnt--;
	lock_release( mt_lock );
	return 1;

error:
	lock_get( mt_lock );
	mt_tree_refcnt--;
	lock_release( mt_lock );
	return -1;
}

static int db_restore(void)
{
	ua_pres_t* p= NULL;
	db_key_t result_cols[20];
	db_res_t *res= NULL;
	db_row_t *row = NULL;
	db_val_t *row_vals= NULL;
	str pres_uri, pres_id, to_uri;
	str etag, tuple_id;
	str watcher_uri, call_id;
	str to_tag, from_tag, remote_contact;
	str record_route, contact, extra_headers;
	int size= 0, i;
	int n_result_cols= 0;
	int puri_col,touri_col,pid_col,expires_col,flag_col,etag_col, desired_expires_col;
	int watcher_col,callid_col,totag_col,fromtag_col,cseq_col,remote_contact_col;
	int event_col,contact_col,tuple_col,record_route_col, extra_headers_col;
	int version_col;
	int no_rows = 10;

	result_cols[puri_col=n_result_cols++]	= &str_pres_uri_col;
	result_cols[touri_col=n_result_cols++]	= &str_to_uri_col;
	result_cols[pid_col=n_result_cols++]	= &str_pres_id_col;
	result_cols[expires_col=n_result_cols++]= &str_expires_col;
	result_cols[flag_col=n_result_cols++]	= &str_flag_col;
	result_cols[etag_col=n_result_cols++]	= &str_etag_col;
	result_cols[tuple_col=n_result_cols++]	= &str_tuple_id_col;
	result_cols[watcher_col=n_result_cols++]= &str_watcher_uri_col;
	result_cols[callid_col=n_result_cols++] = &str_call_id_col;
	result_cols[totag_col=n_result_cols++]	= &str_to_tag_col;
	result_cols[fromtag_col=n_result_cols++]= &str_from_tag_col;
	result_cols[cseq_col= n_result_cols++]	= &str_cseq_col;
	result_cols[event_col= n_result_cols++]	= &str_event_col;
	result_cols[record_route_col= n_result_cols++]	= &str_record_route_col;
	result_cols[contact_col= n_result_cols++]	= &str_contact_col;
	result_cols[remote_contact_col= n_result_cols++]	= &str_remote_contact_col;
	result_cols[extra_headers_col= n_result_cols++]	= &str_extra_headers_col;
	result_cols[desired_expires_col= n_result_cols++]	= &str_desired_expires_col;
	result_cols[version_col= n_result_cols++]	= &str_version_col;

	if(!pua_db)
	{
		LM_ERR("null database connection\n");
		return -1;
	}

	if(pua_dbf.use_table(pua_db, &db_table)< 0)
	{
		LM_ERR("in use table\n");
		return -1;
	}

	if (DB_CAPABILITY(pua_dbf, DB_CAP_FETCH)) {
		if(pua_dbf.query(pua_db,0, 0, 0, result_cols,0, n_result_cols, 0,0)< 0)
		{
			LM_ERR("while querying table\n");
			return -1;
		}
		no_rows = estimate_available_rows( 128+128+8+8+4+32+64+64+128+
			128+64+64+16+64, n_result_cols);
		if (no_rows==0) no_rows = 10;

		if(pua_dbf.fetch_result(pua_db, &res, no_rows)<0)
		{
			LM_ERR("Error fetching rows\n");
			return -1;
		}
	} else {
		if(pua_dbf.query(pua_db,0, 0, 0,result_cols,0,n_result_cols,0,&res)< 0)
		{
			LM_ERR("while querrying table\n");
			if(res)
			{
				pua_dbf.free_result(pua_db, res);
				res = NULL;
			}
			return -1;
		}
	}

	if(res== NULL)
		return -1;

	if(res->n<=0)
	{
		LM_INFO("the query returned no result\n");
		pua_dbf.free_result(pua_db, res);
		res = NULL;
		return 0;
	}

	LM_DBG("found %d db entries\n", res->n);

	do {
		for(i =0 ; i< res->n ; i++)
		{
			row = &res->rows[i];
			row_vals = ROW_VALUES(row);
			if(row_vals[expires_col].val.int_val < time(NULL))
				continue;
//.........这里部分代码省略.........

//.........这里部分代码省略.........

    if (!ro_session) {
        LM_ERR("Can't find a session. This is bad");
        return;
    }

    LM_DBG("event-type=%d", ro_session->event_type);

    //	if (!ro_session->active) {
    //		LM_ALERT("Looks like this session was terminated while requesting more units");
    //		goto exit;
    //		return;
    //	}


    if(ro_session->is_final_allocation) {
        now = get_current_time_micro();
        used_secs = now - ro_session->last_event_timestamp;
        if((ro_session->reserved_secs - used_secs) > 0) {
            update_ro_timer(&ro_session->ro_tl, (ro_session->reserved_secs - used_secs));
            return;
        }
        else {
            ro_session->event_type = no_more_credit;
        }
    }

    switch (ro_session->event_type) {
        case answered:
            now = get_current_time_micro();
            used_secs = rint((now - ro_session->last_event_timestamp) / (float) 1000000);
            call_time = rint((now - ro_session->start_time) / (float) 1000000);

			if ((used_secs + ro_session->billed) < (call_time)) {
				adjustment = call_time - (used_secs + ro_session->billed);
				LM_DBG("Making adjustment for Ro interim timer by adding %d seconds\n", adjustment);
				used_secs += adjustment;
			}
			
            counter_add(ims_charging_cnts_h.billed_secs, used_secs);

            if (ro_session->callid.s != NULL && ro_session->ro_session_id.s != NULL) {
                LM_DBG("Found a session to re-apply for timing [%.*s] and user is [%.*s]\n",
                        ro_session->ro_session_id.len,
                        ro_session->ro_session_id.s,
                        ro_session->asserted_identity.len,
                        ro_session->asserted_identity.s);

                LM_DBG("Call session has been active for %i seconds. The last reserved secs was [%i] and the last event was [%i seconds] ago",
                        (unsigned int) call_time,
                        (unsigned int) ro_session->reserved_secs,
                        (unsigned int) used_secs);

                LM_DBG("Call session [p=%p]: we will now make a request for another [%i] of credit with a usage of [%i] seconds from the last bundle.\n",
                        ro_session,
                        interim_request_credits/* new reservation request amount */,
                        (unsigned int) used_secs/* charged seconds from previous reservation */);

                // Apply for more credit.
                //
                // The function call will return immediately and we will receive the reply asynchronously via a callback
				ro_session->billed += used_secs;
                send_ccr_interim(ro_session, (unsigned int) used_secs, interim_request_credits);
                return;
            } else {
                LM_ERR("Hmmm, the session we have either doesn't have all the data or something else has gone wrong.\n");
                /* put the timer back so the call will be killed according to previous timeout. */
                ro_session->event_type = unknown_error;
                int ret = insert_ro_timer(&ro_session->ro_tl,
                        (ro_session->reserved_secs - used_secs) / 1000000);
                if (ret != 0) {
                    LM_CRIT("unable to insert timer for Ro Session [%.*s]\n",
                            ro_session->ro_session_id.len, ro_session->ro_session_id.s);
                } else {
                    ref_ro_session(ro_session, 1, 0);
                    return;
                }
                LM_ERR("Immediately killing call due to unknown error\n");
            }
            break;
        case delayed_delete:
            destroy_ro_session(ro_session);
            return;
        case pending:
            /* call is not answered yet. No point asking more credit. Just wait for dialog to progress somehow */
            return;
        default:
            LM_ERR("Diameter call session - event [%d]\n", ro_session->event_type);

            if (ro_session->event_type == no_more_credit)
                LM_INFO("Call/session must be ended - no more funds.\n");
            else if (ro_session->event_type == unknown_error)
                LM_ERR("last event caused an error. We will now tear down this session.\n");
    }

    counter_inc(ims_charging_cnts_h.killed_calls);

    dlgb.lookup_terminate_dlg(ro_session->dlg_h_entry, ro_session->dlg_h_id, &default_out_of_credit_hdrs);
    return;
}

//.........这里部分代码省略.........
						I_Disc(p);
					}
					break;
				case Timeout:
					Error(p,p->I_sock);
					p->state = Closed;
				default:
					LM_ERR("sm_process(): In state %s invalid event %s\n",
							dp_states[p->state],dp_events[event-101]);
					goto error;
			}
			break;

		case Wait_I_CEA:
			switch(event){
				case I_Rcv_CEA:
					result_code = Process_CEA(p,msg);
					if (result_code>=2000 && result_code<3000)
						p->state = I_Open;
					else {
						Cleanup(p,p->I_sock);
						p->state = Closed;
					}
					log_peer_list(L_INFO);
					break;
				case R_Conn_CER:
					if (p->r_cer) AAAFreeMessage(&(p->r_cer));
					R_Accept(p,sock);
					result_code = Process_CER(p,msg);
					if (result_code>=2000 && result_code<3000){
						p->state = Wait_Returns;
						if (Elect(p,msg)){
							// won the election = > I_Disc(), R_Send_CEA()
							LM_INFO("sm_process():Wait_I_CEA Win Elect \n");
							sm_process(p,Win_Election,msg,1,sock);
						} else {
							// lost the election => wait for I_Recv_CEA, then R_Disc()
							LM_INFO("sm_process():Wait_I_CEA Lose Elect \n");
							p->r_cer = msg;
							sm_process(p,I_Peer_Disc,0,1,p->I_sock);
						}
					}
					else{
						Snd_CEA(p,msg,result_code,p->R_sock);
						R_Disc(p);
						I_Disc(p);
						p->state=Closed;
						break;
					}
					break;
				case I_Peer_Disc:
					I_Disc(p);
					p->state = Closed;
					break;
				case I_Rcv_Non_CEA:
					Error(p,p->I_sock);
					p->state = Closed;
					break;
				case Timeout:
					Error(p,p->I_sock);
					p->state = Closed;
					break;
				default:
					LM_ERR("sm_process(): In state %s invalid event %s\n",
							dp_states[p->state],dp_events[event-101]);
					goto error;

int h350_call_preferences(struct sip_msg* _msg, pv_elem_t* _avp_name_prefix)
{
	int           rc, i, avp_count = 0;
	struct berval **attr_vals;
	size_t        nmatch = 5;
	regmatch_t    pmatch[5];
	int       avp_name;
	int_str   avp_val;
	str           avp_val_str, avp_name_str,
	              avp_name_prefix_str, call_pref_timeout_str;
	int           call_pref_timeout;
	static char   call_pref_avp_name[AVP_NAME_STR_BUF_LEN];

        /*
         * get avp_name_prefix_str
         */
        if (pv_printf_s(_msg, _avp_name_prefix, &avp_name_prefix_str) != 0)
        {
                LM_ERR("pv_printf_s failed\n");
                return E_H350_INTERNAL;
        }


	/*
	 * get LDAP attribute values
	 */
	if ((rc = ldap_api.ldap_result_attr_vals(
			&h350_call_pref_name, &attr_vals)) < 0)
	{
		LM_ERR("Getting LDAP attribute values failed\n");
		return E_H350_INTERNAL;
	}

	if (rc > 0)
	{
		/* no LDAP values found */
		return E_H350_NO_SUCCESS;
	}

	/*
	 * loop through call pref values and add AVP(s)
	 */

	/* copy avp name prefix into call_pref_avp_name */
	if (avp_name_prefix_str.len < AVP_NAME_STR_BUF_LEN)
	{
		memcpy(call_pref_avp_name, avp_name_prefix_str.s, avp_name_prefix_str.len);
	} else
	{
		LM_ERR("AVP name prefix too long [%d] (max [%d])",
			avp_name_prefix_str.len,
			AVP_NAME_STR_BUF_LEN);
		return E_H350_INTERNAL;
	}

	for (i = 0; attr_vals[i] != NULL; i++)
	{
		if ((rc = regexec(call_pref_preg, attr_vals[i]->bv_val, nmatch, pmatch, 0)) != 0)
		{
			switch (rc)
			{
			case REG_NOMATCH:
				LM_INFO("no h350 call preference regex match for [%s]\n",
						attr_vals[i]->bv_val);
				continue;
			case REG_ESPACE:
				LM_ERR("regexec returned REG_ESPACE - out of memory\n");
			default:
				LM_ERR("regexec failed\n");
				ldap_api.ldap_value_free_len(attr_vals);
				return E_H350_INTERNAL;
			}
		}

		/* calculate call preference sip uri */
		if (avp_name_prefix_str.len + pmatch[2].rm_eo - pmatch[2].rm_so
			>= AVP_NAME_STR_BUF_LEN)
		{
			LM_ERR("AVP name too long for [%s]", attr_vals[i]->bv_val);
			continue;
		}
		avp_val_str.s = attr_vals[i]->bv_val + pmatch[1].rm_so;
		avp_val_str.len = pmatch[1].rm_eo - pmatch[1].rm_so;

		avp_val.s = avp_val_str;

		/* calculate call preference avp name */
		memcpy(	call_pref_avp_name + avp_name_prefix_str.len,
			attr_vals[i]->bv_val + pmatch[2].rm_so,
			pmatch[2].rm_eo - pmatch[2].rm_so);

		avp_name_str.s = call_pref_avp_name;
		avp_name_str.len = avp_name_prefix_str.len + pmatch[2].rm_eo - pmatch[2].rm_so;

		avp_name = get_avp_id(&avp_name_str);
		if (avp_name <= 0) {
			LM_ERR("cannot get avp id\n");
			continue;
		}

//.........这里部分代码省略.........

//.........这里部分代码省略.........
		LOG(L_WARN, "WARNING:qm_free: free(0) called from %s: %s(%d)\n", file, func, line);
#else
		LOG(L_WARN, "WARNING:qm_free: free(0) called\n");
#endif
		return;
	}

#ifdef DBG_QM_MALLOC
	if (p>(void*)qm->last_frag_end || p<(void*)qm->first_frag){
		LOG(L_CRIT, "BUG: qm_free: bad pointer %p (out of memory block!)"
				" called from %s: %s(%d) - aborting\n", p, file, func, line);
		if(likely(cfg_get(core, core_cfg, mem_safety)==0))
			abort();
		else return;
	}
#endif

	f=(struct qm_frag*) ((char*)p-sizeof(struct qm_frag));

#ifdef DBG_QM_MALLOC
	qm_debug_frag(qm, f);
	if (f->u.is_free){
		LOG(L_CRIT, "BUG: qm_free: freeing already freed pointer (%p),"
				" called from %s: %s(%d), first free %s: %s(%ld) - aborting\n",
				p, file, func, line, f->file, f->func, f->line);
		if(likely(cfg_get(core, core_cfg, mem_safety)==0))
			abort();
		else return;
	}
	MDBG("qm_free: freeing frag. %p alloc'ed from %s: %s(%ld)\n",
			f, f->file, f->func, f->line);
#endif
	if (unlikely(f->u.is_free)){
		LM_INFO("freeing a free fragment (%p/%p) - ignore\n",
				f, p);
		return;
	}

	size=f->size;
	qm->used-=size;
	qm->real_used-=size;

#ifdef MEM_JOIN_FREE
	if(unlikely(cfg_get(core, core_cfg, mem_join)!=0)) {
		next=prev=0;
		/* mark this fragment as used (might fall into the middle of joined frags)
		  to give us an extra chance of detecting a double free call (if the joined
		  fragment has not yet been reused) */
		f->u.nxt_free=(void*)0x1L; /* bogus value, just to mark it as free */
		/* join packets if possible*/
		next=FRAG_NEXT(f);
		if (((char*)next < (char*)qm->last_frag_end) && (next->u.is_free)){
			/* join next packet */
#ifdef DBG_QM_MALLOC
			qm_debug_frag(qm, next);
#endif
			qm_detach_free(qm, next);
			size+=next->size+FRAG_OVERHEAD;
			qm->real_used-=FRAG_OVERHEAD;
			qm->free_hash[GET_HASH(next->size)].no--; /* FIXME slow */
			qm->ffrags--;
		}
	
		if (f > qm->first_frag){
			prev=FRAG_PREV(f);
			/*	(struct qm_frag*)((char*)f - (struct qm_frag_end*)((char*)f-

/**
 * params:
 *   the filedes where the data was received.
 * returns:
 *   0 if this fd should be taken out of the poll_fd array bcause it already has AS name.
 *   fd if an AS was completed (returns the fd of the events socket)
 *   -1 if there was an error
 *   -2 if client disconnected and should be closed and taken outside the poll_fd array
 */
static int handle_unc_as_data(int fd)
{
   int i,j,k,len;
   char *name1;
   struct as_entry *as;
   /*first, we see if the data to read is from any of the uncompleted as's*/
   for(i=0;i<2*MAX_UNC_AS_NR ;i++)
      if(unc_as_t[i].valid && unc_as_t[i].fd==fd)
	 break;
   if(i==2*MAX_UNC_AS_NR){
      LM_ERR("has received an fd which is not in uncompleted AS array\n");
      return -1;
   }
   if(unc_as_t[i].flags & HAS_NAME){/*shouldn't happen, if it has a name, it shouldnt be in fdset[]*/
      LM_WARN("this shouldn't happen\n");
      return 0;/*already have a name, please take me out the uncompleted AS array*/
   }
   LM_DBG("Reading client name\n");

   if(-1==(len=read_name(fd,unc_as_t[i].name,MAX_AS_NAME))){
      /*this guy should be disconnected, it sent an AS_NAME too long*/
      LM_ERR("Bad name passed from fd\n");
      unc_as_t[i].valid=0;
      unc_as_t[i].flags=0;
      return -2;
   }else if(len==-2){
      LM_WARN("client disconnected\n");
      return -2;
   }
   name1=unc_as_t[i].name;

   /* Check the name isn't already taken */
   for(as=as_list;as;as=as->next){
      if(as->name.len==len && !memcmp(name1,as->name.s,len)){
	 if(as->connected){
	    LM_WARN("AppServer trying to connect with a name already taken (%.*s)\n",len,name1);
	    unc_as_t[i].valid=0;
	    unc_as_t[i].flags=0;
	    return -2;
	 }
	 break;
      }
   }
   if (!as) {
      LM_ERR("a client tried to connect which is not declared in config. script(%.*s)\n",len,name1);
      unc_as_t[i].valid=0;
      unc_as_t[i].flags=0;
      return -2;
   }
   unc_as_t[i].flags |= HAS_NAME;
   /* the loop's upper bound, 
    * if 'i' is in the lower part, then look for an unc_as in the upper part*/
   k=(i>=MAX_UNC_AS_NR?MAX_UNC_AS_NR:2*MAX_UNC_AS_NR);
   /* the loop's lower bound */
   for(j=(i>=MAX_UNC_AS_NR?0:MAX_UNC_AS_NR);j<k;j++)
      if(unc_as_t[j].valid && 
	    (unc_as_t[j].flags & HAS_NAME) && 
	    !strcmp(unc_as_t[i].name,unc_as_t[j].name))
	 break;
   LM_INFO("Fantastic, we have a new client: %s\n",unc_as_t[i].name);
   if(j==k)/* the unc_as peer's socket hasn't been found, just take this one out of fdset because it already has its name */
      return 0;/*take me out from fdset[]*/
   LM_INFO("EUREKA, we have a new completed AS: %s\n",unc_as_t[i].name);
   /* EUREKA ! we have a sweet pair of AS sockets, with the same name !!*/
   if(add_new_as(i<j?i:j,i<j?j:i,as)==-1){
      close(unc_as_t[j].fd);
      close(unc_as_t[i].fd);
      unc_as_t[j].valid=unc_as_t[i].valid=0;
      unc_as_t[j].flags=unc_as_t[i].flags=0;
      return -1;
   }
   unc_as_t[j].valid=unc_as_t[i].valid=0;
   unc_as_t[j].flags=unc_as_t[i].flags=0;
   return unc_as_t[i<j?i:j].fd;
}

void replicate_ucontact_update(urecord_t *r, ucontact_t *ct)
{
	str st;
	int rc;
	bin_packet_t packet;

	if (bin_init(&packet, &contact_repl_cap, REPL_UCONTACT_UPDATE, BIN_VERSION, 0) != 0) {
		LM_ERR("failed to replicate this event\n");
		return;
	}

	bin_push_str(&packet, r->domain);
	bin_push_str(&packet, &r->aor);
	bin_push_str(&packet, &ct->c);
	bin_push_str(&packet, &ct->callid);
	bin_push_str(&packet, &ct->user_agent);
	bin_push_str(&packet, &ct->path);
	bin_push_str(&packet, &ct->attr);
	bin_push_str(&packet, &ct->received);
	bin_push_str(&packet, &ct->instance);

	st.s = (char *) &ct->expires;
	st.len = sizeof ct->expires;
	bin_push_str(&packet, &st);

	st.s = (char *) &ct->q;
	st.len = sizeof ct->q;
	bin_push_str(&packet, &st);

	bin_push_str(&packet, ct->sock?&ct->sock->sock_str:NULL);
	bin_push_int(&packet, ct->cseq);
	bin_push_int(&packet, ct->flags);
	bin_push_int(&packet, ct->cflags);
	bin_push_int(&packet, ct->methods);

	st.s   = (char *)&ct->last_modified;
	st.len = sizeof ct->last_modified;
	bin_push_str(&packet, &st);

	st = store_serialize(ct->kv_storage);
	if (ZSTR(st))
		LM_ERR("oom\n");
	bin_push_str(&packet, &st);
	store_free_buffer(&st);

	if (cluster_mode == CM_FEDERATION_CACHEDB)
		rc = clusterer_api.send_all_having(&packet, location_cluster,
		                                   NODE_CMP_EQ_SIP_ADDR);
	else
		rc = clusterer_api.send_all(&packet, location_cluster);
	switch (rc) {
	case CLUSTERER_CURR_DISABLED:
		LM_INFO("Current node is disabled in cluster: %d\n", location_cluster);
		goto error;
	case CLUSTERER_DEST_DOWN:
		LM_INFO("All destinations in cluster: %d are down or probing\n",
			location_cluster);
		goto error;
	case CLUSTERER_SEND_ERR:
		LM_ERR("Error sending in cluster: %d\n", location_cluster);
		goto error;
	}

	bin_free_packet(&packet);
	return;

error:
	LM_ERR("replicate ucontact update failed\n");
	bin_free_packet(&packet);
}

static int receive_ucontact_insert(bin_packet_t *packet)
{
	static ucontact_info_t ci;
	static str d, aor, host, contact_str, callid,
		user_agent, path, attr, st, sock;
	udomain_t *domain;
	urecord_t *record;
	ucontact_t *contact;
	int rc, port, proto;

	memset(&ci, 0, sizeof ci);

	bin_pop_str(packet, &d);
	bin_pop_str(packet, &aor);

	if (find_domain(&d, &domain) != 0) {
		LM_ERR("domain '%.*s' is not local\n", d.len, d.s);
		goto error;
	}

	bin_pop_str(packet, &contact_str);

	bin_pop_str(packet, &st);
	memcpy(&ci.contact_id, st.s, sizeof ci.contact_id);

	bin_pop_str(packet, &callid);
	ci.callid = &callid;

	bin_pop_str(packet, &user_agent);
	ci.user_agent = &user_agent;

	bin_pop_str(packet, &path);
	ci.path = &path;

	bin_pop_str(packet, &attr);
	ci.attr = &attr;

	bin_pop_str(packet, &ci.received);
	bin_pop_str(packet, &ci.instance);

	bin_pop_str(packet, &st);
	memcpy(&ci.expires, st.s, sizeof ci.expires);

	bin_pop_str(packet, &st);
	memcpy(&ci.q, st.s, sizeof ci.q);

	bin_pop_str(packet, &sock);

	if (sock.s && sock.s[0]) {
		if (parse_phostport(sock.s, sock.len, &host.s, &host.len,
			&port, &proto) != 0) {
			LM_ERR("bad socket <%.*s>\n", sock.len, sock.s);
			goto error;
		}

		ci.sock = grep_sock_info(&host, (unsigned short) port,
			(unsigned short) proto);
		if (!ci.sock)
			LM_DBG("non-local socket <%.*s>\n", sock.len, sock.s);
	} else {
		ci.sock =  NULL;
	}

	bin_pop_int(packet, &ci.cseq);
	bin_pop_int(packet, &ci.flags);
	bin_pop_int(packet, &ci.cflags);
	bin_pop_int(packet, &ci.methods);

	bin_pop_str(packet, &st);
	memcpy(&ci.last_modified, st.s, sizeof ci.last_modified);

	if (skip_replicated_db_ops)
		ci.flags |= FL_MEM;

	lock_udomain(domain, &aor);

	if (get_urecord(domain, &aor, &record) != 0) {
		LM_INFO("failed to fetch local urecord - creating new one "
			"(ci: '%.*s') \n", callid.len, callid.s);

		if (insert_urecord(domain, &aor, &record, 1) != 0) {
			LM_ERR("failed to insert new record\n");
			unlock_udomain(domain, &aor);
			goto error;
		}
	}

	rc = get_ucontact(record, &contact_str, &callid, ci.cseq, &contact);
	switch (rc) {
	case -2:
		/* received data is consistent with what we have */
	case -1:
		/* received data is older than what we have */
		break;
	case 0:
		/* received data is newer than what we have */
		if (update_ucontact(record, contact, &ci, 1) != 0) {
			LM_ERR("failed to update ucontact (ci: '%.*s')\n", callid.len, callid.s);
			unlock_udomain(domain, &aor);
			goto error;
//.........这里部分代码省略.........

//.........这里部分代码省略.........
	}
	num_pcres_tmp = i + 1;
	
	fclose(f);
	
	/* Fix the patterns */
	for (i=0; i < num_pcres_tmp; i++) {
		
		/* Convert empty groups in unmatcheable regular expression ^$ */
		if (strlen(patterns[i]) == 1) {
			patterns[i][0] = '^';
			patterns[i][1] = '$';
			patterns[i][2] = '\0';
			continue;
		}
		
		/* Delete possible '\n' at the end of the pattern */
		if (patterns[i][strlen(patterns[i])-1] == '\n') {
			patterns[i][strlen(patterns[i])-1] = '\0';
		}
		
		/* Replace '\n' with '|' (except at the end of the pattern) */
		for (j=0; j < strlen(patterns[i]); j++) {
			if (patterns[i][j] == '\n' && j != strlen(patterns[i])-1) {
				patterns[i][j] = '|';
			}
		}
		
		/* Add ')' at the end of the pattern */
		patterns[i][strlen(patterns[i])] = ')';
	}
	
	/* Log the group patterns */
	LM_INFO("num groups = %d\n", num_pcres_tmp);
	for (i=0; i < num_pcres_tmp; i++) {
		LM_INFO("<group[%d]>%s</group[%d]> (size = %i)\n", i, patterns[i], i, (int)strlen(patterns[i]));
	}
	
	/* Temporal pointer of pcres */
	if ((pcres_tmp = pkg_malloc(sizeof(pcre *) * num_pcres_tmp)) == 0) {
		LM_ERR("no more memory for pcres_tmp\n");
		goto err;
	}
	for (i=0; i<num_pcres_tmp; i++) {
		pcres_tmp[i] = NULL;
	}
	
	/* Compile the patters */
	for (i=0; i<num_pcres_tmp; i++) {
	
		pcre_tmp = pcre_compile(patterns[i], pcre_options, &pcre_error, &pcre_erroffset, NULL);
		if (pcre_tmp == NULL) {
			LM_ERR("pcre_tmp compilation of '%s' failed at offset %d: %s\n", patterns[i], pcre_erroffset, pcre_error);
			goto err;
		}
		pcre_rc = pcre_fullinfo(pcre_tmp, NULL, PCRE_INFO_SIZE, &pcre_size);
		if (pcre_rc) {
			printf("pcre_fullinfo on compiled pattern[%i] yielded error: %d\n", i, pcre_rc);
			goto err;
		}
		
		if ((pcres_tmp[i] = pkg_malloc(pcre_size)) == 0) {
			LM_ERR("no more memory for pcres_tmp[%i]\n", i);
			goto err;
		}
		

/**
 * init module function
 */
static int mod_init(void)
{
	sip_uri_t puri;
	char buri[MAX_URI_SIZE];

	if(th_sanity_checks!=0)
	{
		if(sanity_load_api(&scb)<0)
		{
			LM_ERR("cannot bind to sanity module\n");
			goto error;
		}
	}
	if(th_ip.len<=0)
	{
		LM_ERR("mask IP parameter is invalid\n");
		goto error;
	}

	if(th_ip.len + 32 >= MAX_URI_SIZE) {
		LM_ERR("mask address is too long\n");
		goto error;
	}
	memcpy(buri, "sip:", 4);
	memcpy(buri+4, th_ip.s, th_ip.len);
	buri[th_ip.len+8] = '\0';

	if(parse_uri(buri, th_ip.len+4, &puri)<0) {
		LM_ERR("mask uri is invalid\n");
		goto error;
	}
	if(check_self(&puri.host, puri.port_no, 0)==1)
	{
		th_mask_addr_myself = 1;
		LM_INFO("mask address matches myself [%.*s]\n",
				th_ip.len, th_ip.s);
	}

	/* 'SIP/2.0/UDP ' + ip + ';' + param + '=' + prefix (+ '\0') */
	th_via_prefix.len = 12 + th_ip.len + 1 + th_vparam_name.len + 1
		+ th_vparam_prefix.len;
	th_via_prefix.s = (char*)pkg_malloc(th_via_prefix.len+1);
	if(th_via_prefix.s==NULL)
	{
		LM_ERR("via prefix parameter is invalid\n");
		goto error;
	}
	/* 'sip:' + ip + ';' + param + '=' + prefix (+ '\0') */
	th_uri_prefix.len = 4 + th_ip.len + 1 + th_uparam_name.len + 1
		+ th_uparam_prefix.len;
	th_uri_prefix.s = (char*)pkg_malloc(th_uri_prefix.len+1);
	if(th_uri_prefix.s==NULL)
	{
		LM_ERR("uri prefix parameter is invalid\n");
		goto error;
	}
	/* build via prefix */
	memcpy(th_via_prefix.s, "SIP/2.0/UDP ", 12);
	memcpy(th_via_prefix.s+12, th_ip.s, th_ip.len);
	th_via_prefix.s[12+th_ip.len] = ';';
	memcpy(th_via_prefix.s+12+th_ip.len+1, th_vparam_name.s,
			th_vparam_name.len);
	th_via_prefix.s[12+th_ip.len+1+th_vparam_name.len] = '=';
	memcpy(th_via_prefix.s+12+th_ip.len+1+th_vparam_name.len+1,
			th_vparam_prefix.s, th_vparam_prefix.len);
	th_via_prefix.s[th_via_prefix.len] = '\0';
	LM_DBG("VIA prefix: [%s]\n", th_via_prefix.s);
	/* build uri prefix */
	memcpy(th_uri_prefix.s, "sip:", 4);
	memcpy(th_uri_prefix.s+4, th_ip.s, th_ip.len);
	th_uri_prefix.s[4+th_ip.len] = ';';
	memcpy(th_uri_prefix.s+4+th_ip.len+1, th_uparam_name.s, th_uparam_name.len);
	th_uri_prefix.s[4+th_ip.len+1+th_uparam_name.len] = '=';
	memcpy(th_uri_prefix.s+4+th_ip.len+1+th_uparam_name.len+1,
			th_uparam_prefix.s, th_uparam_prefix.len);
	th_uri_prefix.s[th_uri_prefix.len] = '\0';
	LM_DBG("URI prefix: [%s]\n", th_uri_prefix.s);

	th_mask_init();
	sr_event_register_cb(SREV_NET_DATA_IN, th_msg_received);
	sr_event_register_cb(SREV_NET_DATA_OUT, th_msg_sent);
#ifdef USE_TCP
	tcp_set_clone_rcvbuf(1);
#endif
	return 0;
error:
	return -1;
}

/* read from a socket, an AAA message buffer */
int do_read( int socket, rd_buf_t *p)
{
	unsigned char  *ptr;
	unsigned int   wanted_len, len;
	int n;

	if (p->buf==0)
	{
		wanted_len = sizeof(p->first_4bytes) - p->buf_len;
		ptr = ((unsigned char*)&(p->first_4bytes)) + p->buf_len;
	}
	else
	{
		wanted_len = p->first_4bytes - p->buf_len;
		ptr = p->buf + p->buf_len;
	}

	while( (n=recv( socket, ptr, wanted_len, MSG_DONTWAIT ))>0 ) 
	{
//		LM_DBG("(sock=%d)  -> n=%d (expected=%d)\n", p->sock,n,wanted_len);
		p->buf_len += n;
		if (n<wanted_len)
		{
			//LM_DBG("only %d bytes read from %d expected\n",n,wanted_len);
			wanted_len -= n;
			ptr += n;
		}
		else 
		{
			if (p->buf==0)
			{
				/* I just finished reading the first 4 bytes from msg */
				len = ntohl(p->first_4bytes)&0x00ffffff;
				if (len<AAA_MSG_HDR_SIZE || len>MAX_AAA_MSG_SIZE)
				{
					LM_ERR("(sock=%d): invalid message "
						"length read %u (%x)\n", socket, len, p->first_4bytes);
					goto error;
				}
				//LM_DBG("message length = %d(%x)\n",len,len);
				if ( (p->buf=pkg_malloc(len))==0  )
				{
					LM_ERR("no more pkg memory\n");
					goto error;
				}
				*((unsigned int*)p->buf) = p->first_4bytes;
				p->buf_len = sizeof(p->first_4bytes);
				p->first_4bytes = len;
				/* update the reading position and len */
				ptr = p->buf + p->buf_len;
				wanted_len = p->first_4bytes - p->buf_len;
			}
			else
			{
				/* I finished reading the whole message */
				LM_DBG("(sock=%d): whole message read (len=%d)!\n",
					socket, p->first_4bytes);
				return CONN_SUCCESS;
			}
		}
	}

	if (n==0)
	{
		LM_INFO("(sock=%d): FIN received\n", socket);
		return CONN_CLOSED;
	}
	if ( n==-1 && errno!=EINTR && errno!=EAGAIN )
	{
		LM_ERR("(on sock=%d): n=%d , errno=%d (%s)\n",
			socket, n, errno, strerror(errno));
		goto error;
	}
error:
	return CONN_ERROR;
}

/* change the r-uri if it is a PSTN format */
static int prefix2domain(struct sip_msg* msg, int mode, int sd_en)
{
	str *d, p, all={"*",1};
	int plen;
	struct sip_uri uri;

	if(msg==NULL)
	{
		LM_ERR("received null msg\n");
		return -1;
	}

	/* parse the uri, if not yet */
	if(msg->parsed_uri_ok==0)
		if(parse_sip_msg_uri(msg)<0)
		{
			LM_ERR("failed to parse the R-URI\n");
			return -1;
		}

    /* if the user part begin with the prefix for PSTN users, extract the code*/
	if (msg->parsed_uri.user.len<=0)
	{
		LM_DBG("user part of the message is empty\n");
		return -1;
	}

	if(prefix.len>0)
	{
		if (msg->parsed_uri.user.len<=prefix.len)
		{
			LM_DBG("user part is less than prefix\n");
			return -1;
		}
		if(strncasecmp(prefix.s, msg->parsed_uri.user.s, prefix.len)!=0)
		{
			LM_DBG("PSTN prefix did not matched\n");
			return -1;
		}
	}

	if(prefix.len>0 && prefix.len < msg->parsed_uri.user.len
			&& strncasecmp(prefix.s, msg->parsed_uri.user.s, prefix.len)!=0)
	{
		LM_DBG("PSTN prefix did not matched\n");
		return -1;

	}

	p.s   = msg->parsed_uri.user.s + prefix.len;
	p.len = msg->parsed_uri.user.len - prefix.len;

	lock_start_read( pdt_lock );

	if(sd_en==2)
	{
		/* take the domain from  FROM uri as sdomain */
		if(parse_from_header(msg)<0 ||  msg->from == NULL
				|| get_from(msg)==NULL)
		{
			LM_ERR("cannot parse FROM header\n");
			goto error;
		}

		memset(&uri, 0, sizeof(struct sip_uri));
		if (parse_uri(get_from(msg)->uri.s, get_from(msg)->uri.len , &uri)<0)
		{
			LM_ERR("failed to parse From uri\n");
			goto error;
		}

		/* find the domain that corresponds to this prefix */
		plen = 0;
		if((d=pdt_get_domain(*_ptree, &uri.host, &p, &plen))==NULL)
		{
			plen = 0;
			if((d=pdt_get_domain(*_ptree, &all, &p, &plen))==NULL)
			{
				LM_INFO("no prefix found in [%.*s]\n", p.len, p.s);
				goto error;
			}
		}
	} else if(sd_en==1) {
		/* take the domain from  FROM uri as sdomain */
		if(parse_from_header(msg)<0 ||  msg->from == NULL
				|| get_from(msg)==NULL)
		{
			LM_ERR("ERROR cannot parse FROM header\n");
			goto error;
		}

		memset(&uri, 0, sizeof(struct sip_uri));
		if (parse_uri(get_from(msg)->uri.s, get_from(msg)->uri.len , &uri)<0)
		{
			LM_ERR("failed to parse From uri\n");
			goto error;
		}

		/* find the domain that corresponds to this prefix */
//.........这里部分代码省略.........

/**
 * init module function
 */
static int mod_init(void)
{
	LM_INFO("initializing...\n");

	init_db_url( db_url , 0 /*cannot be null*/);
	db_table.len = strlen(db_table.s);
	sdomain_column.len = strlen(sdomain_column.s);
	prefix_column.len = strlen(prefix_column.s);
	domain_column.len = strlen(domain_column.s);
	prefix.len = strlen(prefix.s);

	pdt_char_list.len = strlen(pdt_char_list.s);
	if(pdt_char_list.len<=0)
	{
		LM_ERR("invalid pdt char list\n");
		return -1;
	}
	LM_INFO("pdt_char_list=%s \n",pdt_char_list.s);

	/* binding to mysql module */
	if(db_bind_mod(&db_url, &pdt_dbf))
	{
		LM_ERR("database module not found\n");
		return -1;
	}

	if (!DB_CAPABILITY(pdt_dbf, DB_CAP_ALL))
	{
		LM_ERR("database module does not "
		    "implement all functions needed by the module\n");
		return -1;
	}

	/* open a connection with the database */
	db_con = pdt_dbf.init(&db_url);
	if(db_con==NULL)
	{
		LM_ERR("failed to connect to the database\n");
		return -1;
	}

	if (pdt_dbf.use_table(db_con, &db_table) < 0)
	{
		LM_ERR("failed to use_table\n");
		goto error1;
	}
	LM_DBG("database connection opened successfully\n");

	/* create & init lock */
	if ((pdt_lock = lock_init_rw()) == NULL) {
		LM_CRIT("failed to init lock\n");
		goto error1;
	}

	/* tree pointer in shm */
	_ptree = (pdt_tree_t**)shm_malloc( sizeof(pdt_tree_t*) );
	if (_ptree==0) {
		LM_ERR("out of shm mem for pdtree\n");
		goto error1;
	}
	*_ptree=0;

	/* loading all information from database */
	if(pdt_load_db()!=0)
	{
		LM_ERR("cannot load info from database\n");
		goto error1;
	}

	pdt_dbf.close(db_con);
	db_con = 0;

#if 0
	pdt_print_tree(*_ptree);
#endif

	/* success code */
	return 0;

error1:
	if (pdt_lock)
	{
		lock_destroy_rw( pdt_lock );
		pdt_lock = 0;
	}
	if(_ptree!=0)
		shm_free(_ptree);

	if(db_con!=NULL)
	{
		pdt_dbf.close(db_con);
		db_con = 0;
	}
	return -1;
}

/*
 * main binary packet UDP receiver loop
 */
static void bin_receive_loop(void)
{
	int rcv_bytes;
	struct receive_info ri;
	struct packet_cb_list *p;
	str name;

	ri.bind_address = bind_address;
	ri.dst_port = bind_address->port_no;
	ri.dst_ip = bind_address->address;
	ri.proto = PROTO_UDP;
	ri.proto_reserved1 = ri.proto_reserved2 = 0;

	for (;;) {
		rcv_bytes = recvfrom(bind_address->socket, rcv_buf, BUF_SIZE,
							 0, NULL, NULL);
		if (rcv_bytes == -1) {
			if (errno == EAGAIN) {
				LM_DBG("packet with bad checksum received\n");
				continue;
			}

			LM_ERR("recvfrom: [%d] %s\n", errno, strerror(errno));
			if (errno == EINTR || errno == EWOULDBLOCK || errno == ECONNREFUSED)
				continue;

			return;
		}

		rcv_end = rcv_buf + rcv_bytes;

		if (rcv_bytes < MIN_BIN_PACKET_SIZE) {
			LM_INFO("received invalid packet: len = %d\n", rcv_bytes);
			continue;
		}

		if (!is_valid_bin_packet(rcv_buf)) {
			LM_WARN("Invalid binary packet header! First 10 bytes: %.*s\n",
					10, rcv_buf);
			continue;
		}

		if (!has_valid_checksum(rcv_buf, rcv_bytes)) {
			LM_WARN("binary packet checksum test failed!\n");
			continue;
		}

		get_name(rcv_buf, name);
		cpos = name.s + name.len + CMD_FIELD_SIZE;

		/* packet will be now processed by a specific module */
		for (p = reg_modules; p; p = p->next) {
			if (p->module.len == name.len &&
			    memcmp(name.s, p->module.s, name.len) == 0) {

				LM_DBG("binary Packet CMD: %d. Module: %.*s\n",
						bin_rcv_type, name.len, name.s);

				p->cbf(bin_rcv_type);

				break;
			}
		}
	}
}

static int receive_ucontact_update(bin_packet_t *packet)
{
	static ucontact_info_t ci;
	static str d, aor, host, contact_str, callid,
		user_agent, path, attr, st, sock;
	udomain_t *domain;
	urecord_t *record;
	ucontact_t *contact;
	int port, proto;
	int rc;

	memset(&ci, 0, sizeof ci);

	bin_pop_str(packet, &d);
	bin_pop_str(packet, &aor);

	if (find_domain(&d, &domain) != 0) {
		LM_ERR("domain '%.*s' is not local\n", d.len, d.s);
		goto error;
	}

	bin_pop_str(packet, &contact_str);

	bin_pop_str(packet, &callid);
	ci.callid = &callid;

	bin_pop_str(packet, &user_agent);
	ci.user_agent = &user_agent;

	bin_pop_str(packet, &path);
	ci.path = &path;

	bin_pop_str(packet, &attr);
	ci.attr = &attr;

	bin_pop_str(packet, &ci.received);
	bin_pop_str(packet, &ci.instance);

	bin_pop_str(packet, &st);
	memcpy(&ci.expires, st.s, sizeof ci.expires);

	bin_pop_str(packet, &st);
	memcpy(&ci.q, st.s, sizeof ci.q);

	bin_pop_str(packet, &sock);

	if (sock.s && sock.s[0]) {
		if (parse_phostport(sock.s, sock.len, &host.s, &host.len,
			&port, &proto) != 0) {
			LM_ERR("bad socket <%.*s>\n", sock.len, sock.s);
			goto error;
		}

		ci.sock = grep_sock_info(&host, (unsigned short) port,
			(unsigned short) proto);
		if (!ci.sock)
			LM_DBG("non-local socket <%.*s>\n", sock.len, sock.s);
	} else {
		ci.sock = NULL;
	}

	bin_pop_int(packet, &ci.cseq);
	bin_pop_int(packet, &ci.flags);
	bin_pop_int(packet, &ci.cflags);
	bin_pop_int(packet, &ci.methods);

	bin_pop_str(packet, &st);
	memcpy(&ci.last_modified, st.s, sizeof ci.last_modified);

	bin_pop_str(packet, &st);
	ci.packed_kv_storage = &st;

	if (skip_replicated_db_ops)
		ci.flags |= FL_MEM;

	lock_udomain(domain, &aor);

	/* failure in retrieving a urecord may be ok, because packet order in UDP
	 * is not guaranteed, so update commands may arrive before inserts */
	if (get_urecord(domain, &aor, &record) != 0) {
		LM_INFO("failed to fetch local urecord - create new record and contact"
			" (ci: '%.*s')\n", callid.len, callid.s);

		if (insert_urecord(domain, &aor, &record, 1) != 0) {
			LM_ERR("failed to insert urecord\n");
			unlock_udomain(domain, &aor);
			goto error;
		}

		if (insert_ucontact(record, &contact_str, &ci, &contact, 1) != 0) {
			LM_ERR("failed (ci: '%.*s')\n", callid.len, callid.s);
			unlock_udomain(domain, &aor);
			goto error;
		}
	} else {
		rc = get_ucontact(record, &contact_str, &callid, ci.cseq + 1, &contact);
		if (rc == 1) {
			LM_INFO("contact '%.*s' not found, inserting new (ci: '%.*s')\n",
				contact_str.len, contact_str.s, callid.len, callid.s);

//.........这里部分代码省略.........

/*! Parse the curlcon module parameter
 *
 *	Syntax:
 *		name => proto://user:[email protected]/url/url
 *		name => proto://server/url/url
 *		name => proto://server/url/url;param=value;param=value
 *
 *		the url is very much like CURLs syntax
 *		the url is a base url where you can add local address
 */
int curl_parse_param(char *val)
{
	str name	= STR_NULL;;
	str schema	= STR_NULL;
	str url		= STR_NULL;
	str username	= STR_NULL;
	str password	= STR_NULL;
	str params	= STR_NULL;
	str failover	= STR_NULL;
	unsigned int timeout	= default_connection_timeout;
	str useragent   = { default_useragent, strlen(default_useragent) };
	unsigned int http_follow_redirect = default_http_follow_redirect;

	str i