/* loads data from the db */
table_entry_t* load_info(db_func_t *dr_dbf, db_con_t* db_hdl, str *db_table)
{
	int int_vals[7];
	char *str_vals[2];
	int no_of_results;
	int i, n;
	int no_rows = 5;
	int db_cols = 10;
	unsigned long last_attempt;
	static db_key_t clusterer_machine_id_key = &machine_id_col;
	static db_val_t clusterer_machine_id_value = {
		.type = DB_INT,
		.nul = 0,
	};

	VAL_INT(&clusterer_machine_id_value) = server_id;

	/* the columns from the db table */
	db_key_t columns[10];
	/* result from a db query */
	db_res_t* res;
	/* a row from the db table */
	db_row_t* row;
	/* the processed result */
	table_entry_t *data;

	res = 0;
	data = 0;

	columns[0] = &cluster_id_col;
	columns[1] = &machine_id_col;
	columns[2] = &state_col;
	columns[3] = &description_col;
	columns[4] = &url_col;
	columns[5] = &id_col;
	columns[6] = &last_attempt_col;
	columns[7] = &failed_attempts_col;
	columns[8] = &no_tries_col;
	columns[9] = &duration_col;

	CON_OR_RESET(db_hdl);

	/* checking if the table version is up to date*/
	if (db_check_table_version(dr_dbf, db_hdl, db_table, 1/*version*/) != 0)
		goto error;

	/* read data */
	if (dr_dbf->use_table(db_hdl, db_table) < 0) {
		LM_ERR("cannot select table \"%.*s\"\n", db_table->len, db_table->s);
		goto error;
	}

	LM_DBG("DB query - retrieve the clusters list"
		"in which the specified server runs\n");

	/* first we see in which clusters the specified server runs*/
	if (dr_dbf->query(db_hdl, &clusterer_machine_id_key, &op_eq,
		&clusterer_machine_id_value, columns, 1, 1, 0, &res) < 0) {
		LM_ERR("DB query failed - cannot retrieve the clusters list in which"
			" the specified server runs\n");
		goto error;
	}

	LM_DBG("%d rows found in %.*s\n",
		RES_ROW_N(res), db_table->len, db_table->s);

	if (RES_ROW_N(res) == 0) {
		LM_WARN("No machines found in cluster %d\n", server_id);
		return 0;
	}

	clusterer_cluster_id_key = pkg_realloc(clusterer_cluster_id_key,
		RES_ROW_N(res) * sizeof(db_key_t));
	if (!clusterer_cluster_id_key) {
		LM_ERR("no more pkg memory\n");
		goto error;
	}

	for (i = 0; i < RES_ROW_N(res); i++)
		clusterer_cluster_id_key[i] = &cluster_id_col;

	clusterer_cluster_id_value = pkg_realloc(clusterer_cluster_id_value,
		RES_ROW_N(res) * sizeof(db_val_t));

	if (!clusterer_cluster_id_value) {
		LM_ERR("no more pkg memory\n");
		goto error;
	}

	for (i = 0; i < RES_ROW_N(res); i++) {
		VAL_TYPE(clusterer_cluster_id_value + i) = DB_INT;
		VAL_NULL(clusterer_cluster_id_value + i) = 0;
	}

	for (i = 0; i < RES_ROW_N(res); i++) {
		row = RES_ROWS(res) + i;

		check_val(cluster_id_col, ROW_VALUES(row), DB_INT, 1, 0);
		VAL_INT(clusterer_cluster_id_value + i) = VAL_INT(ROW_VALUES(row));
//.........这里部分代码省略.........

static void trace_onreply_in(struct cell* t, int type, struct tmcb_params *ps)
{
	db_key_t db_keys[NR_KEYS];
	db_val_t db_vals[NR_KEYS];

	static char fromip_buff[IP_ADDR_MAX_STR_SIZE+12];
	static char toip_buff[IP_ADDR_MAX_STR_SIZE+12];
	struct sip_msg* msg;
	struct sip_msg* req;
	int_str        avp_value;
	struct usr_avp *avp;
	char statusbuf[8];
	
	if(t==NULL || t->uas.request==0 || ps==NULL)
	{
		LM_DBG("no uas request, local transaction\n");
		return;
	}

	req = ps->req;
	msg = ps->rpl;
	if(msg==NULL || req==NULL)
	{
		LM_DBG("no reply\n");
		return;
	}
	
	avp = NULL;
	if(traced_user_avp.n!=0)
		avp=search_first_avp(traced_user_avp_type, traced_user_avp,
				&avp_value, 0);

	if((avp==NULL) &&  trace_is_off(req))
	{
		LM_DBG("trace off...\n");
		return;
	}

	if(parse_from_header(msg)==-1 || msg->from==NULL || get_from(msg)==NULL)
	{
		LM_ERR("cannot parse FROM header\n");
		goto error;
	}

	if(parse_headers(msg, HDR_CALLID_F, 0)!=0)
	{
		LM_ERR("cannot parse call-id\n");
		return;
	}

	db_keys[0] = msg_column;
	db_vals[0].type = DB_BLOB;
	db_vals[0].nul = 0;
	if(msg->len>0) {
		db_vals[0].val.blob_val.s   = msg->buf;
		db_vals[0].val.blob_val.len = msg->len;
	} else {
		db_vals[0].val.blob_val.s   = "No reply buffer";
		db_vals[0].val.blob_val.len = sizeof("No reply buffer")-1;
	}

	/* check Call-ID header */
	if(msg->callid==NULL || msg->callid->body.s==NULL)
	{
		LM_ERR("cannot find Call-ID header!\n");
		goto error;
	}

	db_keys[1] = callid_column;
	db_vals[1].type = DB_STR;
	db_vals[1].nul = 0;
	db_vals[1].val.str_val.s = msg->callid->body.s;
	db_vals[1].val.str_val.len = msg->callid->body.len;
	
	db_keys[2] = method_column;
	db_vals[2].type = DB_STR;
	db_vals[2].nul = 0;
	db_vals[2].val.str_val.s = t->method.s;
	db_vals[2].val.str_val.len = t->method.len;
		
	db_keys[3] = status_column;
	db_vals[3].type = DB_STRING;
	db_vals[3].nul = 0;
	strcpy(statusbuf, int2str(ps->code, NULL));
	db_vals[3].val.string_val = statusbuf;
		
	db_keys[4] = fromip_column;
	db_vals[4].type = DB_STRING;
	db_vals[4].nul = 0;
	siptrace_copy_proto(msg->rcv.proto, fromip_buff);
	strcat(fromip_buff, ip_addr2a(&msg->rcv.src_ip));
	strcat(fromip_buff,":");
	strcat(fromip_buff, int2str(msg->rcv.src_port, NULL));
	db_vals[4].val.string_val = fromip_buff;
	
	db_keys[5] = toip_column;
	db_vals[5].type = DB_STRING;
	db_vals[5].nul = 0;
	// db_vals[5].val.string_val = ip_addr2a(&msg->rcv.dst_ip);;
	if(trace_local_ip)
//.........这里部分代码省略.........

static int db_restore(void)
{
	ua_pres_t* p= NULL;
	db_key_t result_cols[19]; 
	db1_res_t *res= NULL;
	db_row_t *row = NULL;	
	db_val_t *row_vals= NULL;
	str pres_uri, pres_id;
	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,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;

	if (dbmode==PUA_DB_ONLY)
	{
		LM_ERR( "db_restore shouldn't be called in PUA_DB_ONLY mode\n" );
		return(-1);
	}

	result_cols[puri_col=n_result_cols++]	= &str_pres_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_fetch_query(&pua_dbf, pua_fetch_rows, 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;
	}

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

		for(i =0 ; i< res->n ; i++)
		{
			row = &res->rows[i];
			row_vals = ROW_VALUES(row);

			pres_uri.s= (char*)row_vals[puri_col].val.string_val;
			pres_uri.len = strlen(pres_uri.s);

			LM_DBG("pres_uri= %.*s\n", pres_uri.len, pres_uri.s);

			memset(&etag,			 0, sizeof(str));
			memset(&tuple_id,		 0, sizeof(str));
			memset(&watcher_uri,	 0, sizeof(str));
			memset(&call_id,		 0, sizeof(str));
			memset(&to_tag,			 0, sizeof(str));
			memset(&from_tag,		 0, sizeof(str));
			memset(&record_route,	 0, sizeof(str));
			memset(&pres_id,         0, sizeof(str));
			memset(&contact,         0, sizeof(str));
//.........这里部分代码省略.........

static int mod_init(void) {
	int i;
    startup_time = (int) time(NULL);
    kz_json_escape_char = kz_json_escape_str.s[0];

    if (dbk_node_hostname.s == NULL) {
	LM_ERR("You must set the node_hostname parameter\n");
	return -1;
    }
    dbk_node_hostname.len = strlen(dbk_node_hostname.s);

    dbk_consumer_event_key.len = strlen(dbk_consumer_event_key.s);
   	dbk_consumer_event_subkey.len = strlen(dbk_consumer_event_subkey.s);


   	if(kz_init_avp()) {
   		LM_ERR("Error in avp params\n");
   		return -1;
   	}

    if(!kz_amqp_init()) {
   		return -1;
    }

    if(kz_timer_ms > 0) {
    	kz_timer_tv.tv_usec = (kz_timer_ms % 1000) * 1000;
    	kz_timer_tv.tv_sec = kz_timer_ms / 1000;
    }
    
    if(dbk_pua_mode == 1) {
		kz_db_url.len = kz_db_url.s ? strlen(kz_db_url.s) : 0;
		LM_DBG("db_url=%s/%d/%p\n", ZSW(kz_db_url.s), kz_db_url.len,kz_db_url.s);
		kz_presentity_table.len = strlen(kz_presentity_table.s);

		if(kz_db_url.len > 0) {

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


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

			kz_pa_db = kz_pa_dbf.init(&kz_db_url);
			if (!kz_pa_db)
			{
				LM_ERR("Connection to database failed\n");
				return -1;
			}

			kz_pa_dbf.close(kz_pa_db);
			kz_pa_db = NULL;
		}
    }


    int total_workers = dbk_consumer_workers + (dbk_consumer_processes * kz_server_counter) + 2;

    register_procs(total_workers);
    cfg_register_child(total_workers);

	if (pipe(kz_cmd_pipe_fds) < 0) {
		LM_ERR("cmd pipe() failed\n");
		return -1;
	}

    kz_worker_pipes_fds = (int*) shm_malloc(sizeof(int) * (dbk_consumer_workers) * 2 );
    kz_worker_pipes = (int*) shm_malloc(sizeof(int) * dbk_consumer_workers);
    for(i=0; i < dbk_consumer_workers; i++) {
    	kz_worker_pipes_fds[i*2] = kz_worker_pipes_fds[i*2+1] = -1;
		if (pipe(&kz_worker_pipes_fds[i*2]) < 0) {
			LM_ERR("worker pipe(%d) failed\n", i);
			return -1;
		}
    }

	kz_cmd_pipe = kz_cmd_pipe_fds[1];
	for(i=0; i < dbk_consumer_workers; i++) {
		kz_worker_pipes[i] = kz_worker_pipes_fds[i*2+1];
	}

    return 0;
}

/* synchronize backend with the db */
static void update_db_handler(unsigned int ticks, void* param)
{
	/* data */
	table_entry_t *head_table;
	table_entry_value_t *value;
	table_entry_info_t *info;
	/* columns to be compared ( clusterer_id_col ) */
	db_key_t key_cmp;
	/* with values */
	db_val_t val_cmp;
	/* columns to be set */
	db_key_t key_set[3];
	/* with values */
	db_val_t val_set[3];
	int i;

	CON_OR_RESET(db_hdl);

	/* table to use*/
	if (dr_dbf.use_table(db_hdl, &db_table) < 0) {
		LM_ERR("cannot select table \"%.*s\"\n", db_table.len, db_table.s);
		return;
	}

	val_cmp.type = DB_INT;
	val_cmp.nul = 0;

	for (i = 0; i < 2; ++i) {
		val_set[i].type = DB_INT;
		val_set[i].nul = 0;
	}

	val_set[2].type = DB_BIGINT;
	val_set[2].nul = 0;

	key_cmp = &id_col;

	key_set[0] = &state_col;
	key_set[1] = &no_tries_col;
	key_set[2] = &last_attempt_col;


	lock_start_write(ref_lock);

	head_table = *tdata;
	/* iterating through backend storage to find all data that
	 * must be synchronized with the db */
	while (head_table != NULL) {
		info = head_table->info;
		while (info != NULL) {
			value = info->value;
			while (value != NULL) {
				if (value->dirty_bit == 1) {
					LM_DBG("setting row with primary key %d the status %d\n",
						value->id, value->state);

					val_cmp.val.int_val = value->id;
					val_set[0].val.int_val = value->state;
					val_set[1].val.int_val = value->no_tries;
					val_set[2].val.int_val = value->last_attempt;

					/* updating */
					if (dr_dbf.update(db_hdl, &key_cmp, &op_eq, &val_cmp, key_set, val_set, 1, 3) < 0) {
						LM_ERR("DB update failed\n");
					} else {
						/* only if the query is successful the data is synchronized */
						value->dirty_bit = 0;
					}
				}
				value = value->next;
			}
			info = info->next;
		}
		head_table = head_table->next;
	}

	lock_stop_write(ref_lock);

}

static int mod_init(void)
{
	int heartbeats_timer_interval;
	cluster_info_t *cl;

	LM_INFO("Clusterer module - initializing\n");

	db_table.len = strlen(db_table.s);
	id_col.len = strlen(id_col.s);
	cluster_id_col.len = strlen(cluster_id_col.s);
	node_id_col.len = strlen(node_id_col.s);
	url_col.len = strlen(url_col.s);
	state_col.len = strlen(state_col.s);
	no_ping_retries_col.len = strlen(no_ping_retries_col.s);
	priority_col.len = strlen(priority_col.s);
	sip_addr_col.len = strlen(sip_addr_col.s);
	flags_col.len = strlen(flags_col.s);
	description_col.len = strlen(description_col.s);

	/* only allow the DB URL to be skipped in "P2P discovery" mode */
	init_db_url(clusterer_db_url, db_mode == 0);

	if (current_id < 1) {
		LM_CRIT("Invalid 'my_node_id' parameter\n");
		return -1;
	}
	if (ping_interval <= 0) {
		LM_WARN("Invalid ping_interval parameter, using default value\n");
		ping_interval = DEFAULT_PING_INTERVAL;
	}
	if (node_timeout < 0) {
		LM_WARN("Invalid node_timeout parameter, using default value\n");
		node_timeout = DEFAULT_NODE_TIMEOUT;
	}
	if (ping_timeout <= 0) {
		LM_WARN("Invalid ping_timeout parameter, using default value\n");
		ping_timeout = DEFAULT_PING_TIMEOUT;
	}

	/* create & init lock */
	if ((cl_list_lock = lock_init_rw()) == NULL) {
		LM_CRIT("Failed to init lock\n");
		return -1;
	}

	/* data pointer in shm */
	if (cluster_list == NULL) {
		cluster_list = shm_malloc(sizeof *cluster_list);
		if (!cluster_list) {
			LM_CRIT("No more shm memory\n");
			goto error;
		}
		*cluster_list = NULL;
	} else {
		/* sanity check of my_node_id if node_id also set in a my_node_info param */
		for (cl = *cluster_list; cl; cl = cl->next)
			if (cl->current_node->node_id != current_id) {
				LM_ERR("Bad 'my_node_id' parameter, value: %d different than"
					" the node_id property in the 'my_node_info' parameter\n", current_id);
				goto error;
			}
	}

	if (db_mode) {
		/* bind to the mysql module */
		if (db_bind_mod(&clusterer_db_url, &dr_dbf)) {
			LM_CRIT("Cannot bind to database module! "
				"Did you forget to load a database module ?\n");
			goto error;
		}
		if (!DB_CAPABILITY(dr_dbf, DB_CAP_QUERY)) {
			LM_CRIT("Given SQL DB does not provide query types needed by this module!\n");
			goto error;
		}
		/* init DB connection */
		if ((db_hdl = dr_dbf.init(&clusterer_db_url)) == 0) {
			LM_ERR("cannot initialize database connection\n");
			goto error;
		}
		if (load_db_info(&dr_dbf, db_hdl, &db_table, cluster_list) < 0) {
			LM_ERR("Failed to load info from DB\n");
			goto error;
		}
	}

	/* register timer */
	heartbeats_timer_interval = gcd(ping_interval*1000, ping_timeout);
	heartbeats_timer_interval = gcd(heartbeats_timer_interval, node_timeout*1000);

	if (heartbeats_timer_interval % 1000 == 0) {
		if (register_timer("clstr-heartbeats-timer", heartbeats_timer_handler,
			NULL, heartbeats_timer_interval/1000, TIMER_FLAG_DELAY_ON_DELAY) < 0) {
			LM_CRIT("Unable to register clusterer heartbeats timer\n");
			goto error;
		}
	} else {
		if (register_utimer("clstr-heartbeats-utimer", heartbeats_utimer_handler,
			NULL, heartbeats_timer_interval*1000, TIMER_FLAG_DELAY_ON_DELAY) < 0) {
			LM_CRIT("Unable to register clusterer heartbeats timer\n");
			goto error;
//.........这里部分代码省略.........

/** Module init function */
static int mod_init(void)
{
	char* p = NULL;
	char* flags = NULL;
	int regexp_flags = 0;
	int i = 0, j;
	pv_spec_t avp_spec;

	LM_DBG("start\n");

	/* load b2b_entities api */
	if(load_b2b_api(&b2b_api)< 0)
	{
		LM_ERR("Failed to load b2b api\n");
		return -1;
	}

	if(b2bl_hsize< 1 || b2bl_hsize> 20)
	{
		LM_ERR("Wrong hash size. Needs to be greater than 1"
				" and smaller than 20. Be aware that you should set the log 2"
				" value of the real size\n");
		return -1;
	}
	b2bl_hsize = 1<<b2bl_hsize;

	if(server_address.s == NULL)
	{
		if(extern_scenarios)
		{
			LM_ERR("'server_address' parameter not set. This parameter is"
				" compulsory if you want to use extern scenarios. It must"
				" be set to the IP address of the machine\n");
			return -1;
		}
	}
	else
		server_address.len = strlen(server_address.s);

	if(init_b2bl_htable() < 0)
	{
		LM_ERR("Failed to initialize b2b logic hash table\n");
		return -1;
	}

	if(b2b_clean_period < 0)
	{
		LM_ERR("Wrong parameter - b2b_clean_period [%d]\n", b2b_clean_period);
		return -1;
	}
	if(b2b_update_period < 0)
	{
		LM_ERR("Wrong parameter - b2b_update_period [%d]\n", b2b_update_period);
		return -1;
	}

	if(b2bl_db_mode && db_url.s)
	{
		db_url.len = strlen(db_url.s);
		b2bl_dbtable.len = strlen(b2bl_dbtable.s);
		/* binding to database module  */
		if (db_bind_mod(&db_url, &b2bl_dbf))
		{
			LM_ERR("Database module not found\n");
			return -1;
		}

		if (!DB_CAPABILITY(b2bl_dbf, DB_CAP_ALL))
		{
			LM_ERR("Database module does not implement all functions"
					" needed by b2b_entities module\n");
			return -1;
		}
		b2bl_db = b2bl_dbf.init(&db_url);
		if(!b2bl_db)
		{
			LM_ERR("connecting to database failed\n");
			return -1;
		}

		/*verify table versions */
		if(db_check_table_version(&b2bl_dbf, b2bl_db, &b2bl_dbtable, TABLE_VERSION) < 0)
		{
			LM_ERR("error during table version check\n");
			return -1;
		}

		b2bl_db_init();

		/* reload data */
		if(b2b_logic_restore() < 0)
		{
			LM_ERR("Failed to restore data from database\n");
			return -1;
		}

		if(b2bl_db)
			b2bl_dbf.close(b2bl_db);
		b2bl_db = NULL;
//.........这里部分代码省略.........

static int mod_init(void)
{
	pv_spec_t avp_spec;
	int i;

	init_db_url( db_url , 0 /*cannot be null*/);
	siptrace_table.len = strlen(siptrace_table.s);
	date_column.len = strlen(date_column.s);
	callid_column.len = strlen(callid_column.s);
	traced_user_column.len = strlen(traced_user_column.s);
	msg_column.len = strlen(msg_column.s);
	method_column.len = strlen(method_column.s);
	status_column.len = strlen(status_column.s);
	fromproto_column.len = strlen(fromproto_column.s);
	fromip_column.len = strlen(fromip_column.s);
	fromport_column.len = strlen(fromport_column.s);
	toproto_column.len = strlen(toproto_column.s);
	toip_column.len = strlen(toip_column.s);
	toport_column.len = strlen(toport_column.s);
	fromtag_column.len = strlen(fromtag_column.s);
	direction_column.len = strlen(direction_column.s);
	if (traced_user_avp_str.s)
		traced_user_avp_str.len = strlen(traced_user_avp_str.s);
	if (trace_table_avp_str.s)
		trace_table_avp_str.len = strlen(trace_table_avp_str.s);
	if (dup_uri_str.s)
		dup_uri_str.len = strlen(dup_uri_str.s);

	if (trace_local_ip.s)
		parse_trace_local_ip();

	LM_INFO("initializing...\n");

	fix_flag_name(trace_flag_str, trace_flag);

	trace_flag = get_flag_id_by_name(FLAG_TYPE_MSG, trace_flag_str);

	if (flag_idx2mask(&trace_flag)<0)
		return -1;

	trace_to_database_flag = (int*)shm_malloc(sizeof(int));
        if(trace_to_database_flag==NULL) {
                LM_ERR("no more shm memory left\n");
                return -1;
        }

	*trace_to_database_flag = trace_to_database;

	if(trace_to_database_flag!=NULL && *trace_to_database_flag!=0) {
		/* Find a database module */
		if (db_bind_mod(&db_url, &db_funcs))
		{
			LM_ERR("unable to bind database module\n");
			return -1;
		}
		if (trace_to_database_flag && !DB_CAPABILITY(db_funcs, DB_CAP_INSERT))
		{
			LM_ERR("database modules does not provide all functions needed by module\n");
			return -1;
		}

		if ((db_con = db_funcs.init(&db_url)) == 0) {
			LM_CRIT("Cannot connect to DB\n");
			return -1;
		}

		if(db_check_table_version(&db_funcs, db_con,
				&siptrace_table, SIPTRACE_TABLE_VERSION) < 0) {
			LM_ERR("error during table version check.\n");
			return -1;
		}

		db_funcs.close(db_con);
		db_con = 0;
	}

	trace_on_flag = (int*)shm_malloc(sizeof(int));
	if(trace_on_flag==NULL)
	{
		LM_ERR("no more shm memory left\n");
		return -1;
	}

	*trace_on_flag = trace_on;



	/* register callbacks to TM */
	if (load_tm_api(&tmb)!=0)
	{
		LM_ERR("can't load tm api\n");
		return -1;
	}

	if(tmb.register_tmcb( 0, 0, TMCB_REQUEST_IN, trace_onreq_in, 0, 0) <=0)
	{
		LM_ERR("can't register trace_onreq_in\n");
		return -1;
	}

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

int update_watchers_status(str pres_uri, pres_ev_t* ev, str* rules_doc)
{
	subs_t subs;
	db_key_t query_cols[6], result_cols[5];
	db_val_t query_vals[6];
	int n_result_cols= 0, n_query_cols = 0;
	db1_res_t* result= NULL;
	db_row_t *row;
	db_val_t *row_vals ;
	int i;
	str w_user, w_domain, reason= {0, 0};
	unsigned int status;
	int status_col, w_user_col, w_domain_col, reason_col;
	subs_t* subs_array= NULL,* s;
	unsigned int hash_code;
	int err_ret= -1;
	int n= 0;

	typedef struct ws
	{
		int status;
		str reason;
		str w_user;
		str w_domain;
	}ws_t;
	ws_t* ws_list= NULL;

	LM_DBG("start\n");

	if(ev->content_type.s== NULL)
	{
		ev= contains_event(&ev->name, NULL);
		if(ev== NULL)
		{
			LM_ERR("wrong event parameter\n");
			return 0;
		}
	}

	memset(&subs, 0, sizeof(subs_t));
	subs.pres_uri= pres_uri;
	subs.event= ev;
	subs.auth_rules_doc= rules_doc;

	/* update in watchers_table */
	query_cols[n_query_cols]= &str_presentity_uri_col;
	query_vals[n_query_cols].nul= 0;
	query_vals[n_query_cols].type= DB1_STR;
	query_vals[n_query_cols].val.str_val= pres_uri;
	n_query_cols++;

	query_cols[n_query_cols]= &str_event_col;
	query_vals[n_query_cols].nul= 0;
	query_vals[n_query_cols].type= DB1_STR;
	query_vals[n_query_cols].val.str_val= ev->name;
	n_query_cols++;

	result_cols[status_col= n_result_cols++]= &str_status_col;
	result_cols[reason_col= n_result_cols++]= &str_reason_col;
	result_cols[w_user_col= n_result_cols++]= &str_watcher_username_col;
	result_cols[w_domain_col= n_result_cols++]= &str_watcher_domain_col;

	if (pa_dbf.use_table(pa_db, &watchers_table) < 0) 
	{
		LM_ERR( "in use_table\n");
		goto done;
	}

	if(pa_dbf.query(pa_db, query_cols, 0, query_vals, result_cols,n_query_cols,
				n_result_cols, 0, &result)< 0)
	{
		LM_ERR( "in sql query\n");
		goto done;
	}
	if(result== NULL)
		return 0;

	if(result->n<= 0)
	{
		err_ret= 0;
		goto done;
	}

	LM_DBG("found %d record-uri in watchers_table\n", result->n);
	hash_code= core_case_hash(&pres_uri, &ev->name, shtable_size);
	subs.db_flag= hash_code;

	/*must do a copy as sphere_check requires database queries */
	if(sphere_enable)
	{
        	n= result->n;
		ws_list= (ws_t*)pkg_malloc(n * sizeof(ws_t));
		if(ws_list== NULL)
		{
			LM_ERR("No more private memory\n");
			goto done;
		}
		memset(ws_list, 0, n * sizeof(ws_t));

		for(i= 0; i< result->n ; i++)
//.........这里部分代码省略.........

/*
 * Closes the DB connection.
 */
void matrix_db_close(void) {
	if (matrix_dbh) {
		matrix_dbf.close(matrix_dbh);
		matrix_dbh = NULL;
	}
}

/**
 * Initialize children
 */
static int child_init(int rank)
{
	if (rank==PROC_INIT || rank==PROC_TCP_MAIN)
		return 0;

	pid = my_pid();
	
	if(library_mode)
		return 0;

	if (rank == PROC_MAIN)
	{
		int i;

		for (i = 0; i < pres_notifier_processes; i++)
		{
			char tmp[21];
			snprintf(tmp, 21, "PRESENCE NOTIFIER %d", i);
			pres_notifier_id[i] = i;

			if (fork_basic_utimer(PROC_TIMER, tmp, 1,
						pres_timer_send_notify,
						&pres_notifier_id[i],
						1000000/pres_notifier_poll_rate) < 0)
			{
				LM_ERR("Failed to start PRESENCE NOTIFIER %d\n", i);
				return -1;
			}
		}

		return 0;
	}

	if (pa_dbf.init==0)
	{
		LM_CRIT("child_init: database not bound\n");
		return -1;
	}
	/* Do not pool the connections where possible when running notifier
       processes. */
	if (pres_notifier_processes > 0 && pa_dbf.init2)
		pa_db = pa_dbf.init2(&db_url, DB_POOLING_NONE);
	else
		pa_db = pa_dbf.init(&db_url);
	if (!pa_db)
	{
		LM_ERR("child %d: unsuccessful connecting to database\n", rank);
		return -1;
	}
	
	if (pa_dbf.use_table(pa_db, &presentity_table) < 0)
	{
		LM_ERR( "child %d:unsuccessful use_table presentity_table\n", rank);
		return -1;
	}

	if (pa_dbf.use_table(pa_db, &active_watchers_table) < 0)  
	{
		LM_ERR( "child %d:unsuccessful use_table active_watchers_table\n",
				rank);
		return -1;
	}

	if (pa_dbf.use_table(pa_db, &watchers_table) < 0)  
	{
		LM_ERR( "child %d:unsuccessful use_table watchers_table\n", rank);
		return -1;
	}

	LM_DBG("child %d: Database connection opened successfully\n", rank);
	
	return 0;
}

int pres_update_status(subs_t subs, str reason, db_key_t* query_cols,
        db_val_t* query_vals, int n_query_cols, subs_t** subs_array)
{
	db_key_t update_cols[5];
	db_val_t update_vals[5];
	int n_update_cols= 0;
	int u_status_col, u_reason_col, q_wuser_col, q_wdomain_col;
	int status;
	query_cols[q_wuser_col=n_query_cols]= &str_watcher_username_col;
	query_vals[n_query_cols].nul= 0;
	query_vals[n_query_cols].type= DB1_STR;
	n_query_cols++;

	query_cols[q_wdomain_col=n_query_cols]= &str_watcher_domain_col;
	query_vals[n_query_cols].nul= 0;
	query_vals[n_query_cols].type= DB1_STR;
	n_query_cols++;

	update_cols[u_status_col= n_update_cols]= &str_status_col;
	update_vals[u_status_col].nul= 0;
	update_vals[u_status_col].type= DB1_INT;
	n_update_cols++;

	update_cols[u_reason_col= n_update_cols]= &str_reason_col;
	update_vals[u_reason_col].nul= 0;
	update_vals[u_reason_col].type= DB1_STR;
	n_update_cols++;

	status= subs.status;
	if(subs.event->get_auth_status(&subs)< 0)
	{
		LM_ERR( "getting status from rules document\n");
		return -1;
	}
	LM_DBG("subs.status= %d\n", subs.status);
	if(get_status_str(subs.status)== NULL)
	{
		LM_ERR("wrong status: %d\n", subs.status);
		return -1;
	}

	if(subs.status!= status || reason.len!= subs.reason.len ||
		(reason.s && subs.reason.s && strncmp(reason.s, subs.reason.s,
											  reason.len)))
	{
		/* update in watchers_table */
		query_vals[q_wuser_col].val.str_val= subs.watcher_user; 
		query_vals[q_wdomain_col].val.str_val= subs.watcher_domain; 

		update_vals[u_status_col].val.int_val= subs.status;
		update_vals[u_reason_col].val.str_val= subs.reason;

		if (pa_dbf.use_table(pa_db, &watchers_table) < 0) 
		{
			LM_ERR( "in use_table\n");
			return -1;
		}

		if(pa_dbf.update(pa_db, query_cols, 0, query_vals, update_cols,
					update_vals, n_query_cols, n_update_cols)< 0)
		{
			LM_ERR( "in sql update\n");
			return -1;
		}
		/* save in the list all affected dialogs */
		/* if status switches to terminated -> delete dialog */
		if(update_pw_dialogs(&subs, subs.db_flag, subs_array)< 0)
		{
			LM_ERR( "extracting dialogs from [watcher]=%.*[email protected]%.*s to"
				" [presentity]=%.*s\n",	subs.watcher_user.len, subs.watcher_user.s,
				subs.watcher_domain.len, subs.watcher_domain.s, subs.pres_uri.len,
				subs.pres_uri.s);
			return -1;
		}
	}
    return 0;
}

/**
 * init module function
 */
static int mod_init(void)
{
	if(pres_uri_match == 1) {
		presence_sip_uri_match = sip_uri_case_insensitive_match;
	} else {
		presence_sip_uri_match = sip_uri_case_sensitive_match;
	}

	if(register_mi_mod(exports.name, mi_cmds)!=0)
	{
		LM_ERR("failed to register MI commands\n");
		return -1;
	}
	if(presence_init_rpc()!=0)
	{
		LM_ERR("failed to register RPC commands\n");
		return -1;
	}

	LM_DBG("db_url=%s/%d/%p\n", ZSW(db_url.s), db_url.len,db_url.s);

	if(db_url.s== NULL)
		library_mode= 1;

	EvList= init_evlist();
	if(!EvList){
		LM_ERR("unsuccessful initialize event list\n");
		return -1;
	}

	if(library_mode== 1)
	{
		LM_DBG("Presence module used for API library purpose only\n");
		return 0;
	}

	if(expires_offset<0)
		expires_offset = 0;
	
	if(to_tag_pref==NULL || strlen(to_tag_pref)==0)
		to_tag_pref="10";

	if(max_expires<= 0)
		max_expires = 3600;

	if(min_expires < 0)
		min_expires = 0;

	if(min_expires > max_expires)
		min_expires = max_expires;

    if(min_expires_action < 1 || min_expires_action > 2) {
        LM_ERR("min_expires_action must be 1 = RFC 6665/3261 Reply 423, 2 = force min_expires value\n");
        return -1;
    }
    
	if(server_address.s== NULL)
		LM_DBG("server_address parameter not set in configuration file\n");

	/* bind the SL API */
	if (sl_load_api(&slb)!=0) {
		LM_ERR("cannot bind to SL API\n");
		return -1;
	}

	/* load all TM stuff */
	if(load_tm_api(&tmb)==-1)
	{
		LM_ERR("Can't load tm functions. Module TM not loaded?\n");
		return -1;
	}
	
	if(db_url.s== NULL)
	{
		LM_ERR("database url not set!\n");
		return -1;
	}

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

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

	pa_db = pa_dbf.init(&db_url);
	if (!pa_db)
	{
		LM_ERR("Connection to database failed\n");
//.........这里部分代码省略.........

static int update_pw_dialogs_dbonlymode(subs_t* subs, subs_t** subs_array)
{
	db_key_t query_cols[5], db_cols[3];
	db_val_t query_vals[5], db_vals[3];
	db_key_t result_cols[24];
	int n_query_cols=0, n_result_cols=0, n_update_cols=0;
	int event_col, pres_uri_col, watcher_user_col, watcher_domain_col;
	int r_pres_uri_col,r_to_user_col,r_to_domain_col;
	int r_from_user_col,r_from_domain_col,r_callid_col;
	int r_to_tag_col,r_from_tag_col,r_sockinfo_col;
	int r_event_id_col,r_local_contact_col,r_contact_col;
	int r_record_route_col, r_reason_col;
	int r_event_col, r_local_cseq_col, r_remote_cseq_col;
	int r_status_col, r_version_col;
	int r_expires_col, r_watcher_user_col, r_watcher_domain_col;
	db1_res_t *result= NULL;
 	db_val_t *row_vals;
	db_row_t *rows;
	int nr_rows, loop;
	subs_t s, *cs;
	str ev_sname;

	if(pa_db == NULL)
	{
		LM_ERR("null database connection\n");
		return(-1);
	}

	if (pa_dbf.use_table(pa_db, &active_watchers_table) < 0) 
	{
		LM_ERR("use table failed\n");
		return(-1);
	}

	query_cols[event_col=n_query_cols]= &str_event_col;
	query_vals[event_col].nul= 0;
	query_vals[event_col].type= DB1_STR;
	query_vals[event_col].val.str_val= subs->event->name ;
	n_query_cols++;

	query_cols[pres_uri_col=n_query_cols]= &str_presentity_uri_col;
	query_vals[pres_uri_col].nul= 0;
	query_vals[pres_uri_col].type= DB1_STR;
	query_vals[pres_uri_col].val.str_val= subs->pres_uri;
	n_query_cols++;

	query_cols[watcher_user_col=n_query_cols]= &str_watcher_username_col;
	query_vals[watcher_user_col].nul= 0;
	query_vals[watcher_user_col].type= DB1_STR;
	query_vals[watcher_user_col].val.str_val= subs->watcher_user;
	n_query_cols++;

	query_cols[watcher_domain_col=n_query_cols]= &str_watcher_domain_col;
	query_vals[watcher_domain_col].nul= 0;
	query_vals[watcher_domain_col].type= DB1_STR;
	query_vals[watcher_domain_col].val.str_val= subs->watcher_domain;
	n_query_cols++;


	result_cols[r_to_user_col=n_result_cols++] = &str_to_user_col;
	result_cols[r_to_domain_col=n_result_cols++] = &str_to_domain_col;
	result_cols[r_from_user_col=n_result_cols++] = &str_from_user_col;
	result_cols[r_from_domain_col=n_result_cols++] = &str_from_domain_col;
	result_cols[r_watcher_user_col=n_result_cols++] = &str_watcher_username_col;
	result_cols[r_watcher_domain_col=n_result_cols++] = &str_watcher_domain_col;
	result_cols[r_callid_col=n_result_cols++] = &str_callid_col;
	result_cols[r_to_tag_col=n_result_cols++] = &str_to_tag_col;
	result_cols[r_from_tag_col=n_result_cols++] = &str_from_tag_col;
	result_cols[r_sockinfo_col=n_result_cols++] = &str_socket_info_col;
	result_cols[r_event_id_col=n_result_cols++] = &str_event_id_col;
	result_cols[r_local_contact_col=n_result_cols++] = &str_local_contact_col;
	result_cols[r_record_route_col=n_result_cols++] = &str_record_route_col;
	result_cols[r_reason_col=n_result_cols++] = &str_reason_col;
	result_cols[r_local_cseq_col=n_result_cols++] = &str_local_cseq_col;
	result_cols[r_version_col=n_result_cols++] = &str_version_col;
	result_cols[r_expires_col=n_result_cols++] = &str_expires_col;
	result_cols[r_event_col=n_result_cols++] = &str_event_col;
	result_cols[r_pres_uri_col=n_result_cols++] = &str_presentity_uri_col;
	result_cols[r_contact_col=n_result_cols++] = &str_contact_col;

	/* these ones are unused for some reason !!! */
	result_cols[r_remote_cseq_col=n_result_cols++] = &str_remote_cseq_col;
	result_cols[r_status_col=n_result_cols++] = &str_status_col;
	/*********************************************/

	if(pa_dbf.query(pa_db, query_cols, 0, query_vals, result_cols, 
				n_query_cols, n_result_cols, 0, &result )< 0)
	{
		LM_ERR("Can't query db\n");
		if(result) pa_dbf.free_result(pa_db, result);
		return(-1);
	}

	if(result == NULL) return(-1);

	nr_rows = RES_ROW_N(result);

	LM_DBG("found %d matching dialogs\n", nr_rows);

	if (nr_rows <= 0)
//.........这里部分代码省略.........

static void trace_onreply_out(struct cell* t, int type, struct tmcb_params *ps)
{
	db_key_t db_keys[NR_KEYS];
	db_val_t db_vals[NR_KEYS];
	int faked = 0;
	static char fromip_buff[IP_ADDR_MAX_STR_SIZE+12];
	static char toip_buff[IP_ADDR_MAX_STR_SIZE+12];
	struct sip_msg* msg;
	struct sip_msg* req;
	int_str        avp_value;
	struct usr_avp *avp;
	struct ip_addr to_ip;
	int len;
	char statusbuf[8];
	str *sbuf;
	struct dest_info *dst;

	if (t==NULL || t->uas.request==0 || ps==NULL)
	{
		LM_DBG("no uas request, local transaction\n");
		return;
	}
	
	avp = NULL;
	if(traced_user_avp.n!=0)
		avp=search_first_avp(traced_user_avp_type, traced_user_avp,
				&avp_value, 0);

	if((avp==NULL) &&  trace_is_off(t->uas.request))
	{
		LM_DBG("trace off...\n");
		return;
	}
	
	req = ps->req;
	msg = ps->rpl;
	if(msg==NULL || msg==FAKED_REPLY)
	{
		msg = t->uas.request;
		faked = 1;
	}

	if(parse_from_header(msg)==-1 || msg->from==NULL || get_from(msg)==NULL)
	{
		LM_ERR("cannot parse FROM header\n");
		goto error;
	}

	if(parse_headers(msg, HDR_CALLID_F, 0)!=0)
	{
		LM_ERR("cannot parse call-id\n");
		return;
	}

	db_keys[0] = msg_column;
	db_vals[0].type = DB_BLOB;
	db_vals[0].nul = 0;
	sbuf = (str*)ps->extra1;
	if(faked==0)
	{
		if(sbuf!=0 && sbuf->len>0) {
			db_vals[0].val.blob_val.s   = sbuf->s;
			db_vals[0].val.blob_val.len = sbuf->len;
		} else if(t->uas.response.buffer.s!=NULL) {
			db_vals[0].val.blob_val.s   = t->uas.response.buffer.s;
			db_vals[0].val.blob_val.len = t->uas.response.buffer.len;
		} else if(msg->len>0) {
			db_vals[0].val.blob_val.s   = msg->buf;
			db_vals[0].val.blob_val.len = msg->len;
		} else {
			db_vals[0].val.blob_val.s   = "No reply buffer";
			db_vals[0].val.blob_val.len = sizeof("No reply buffer")-1;
		}
	} else {
		if(sbuf!=0 && sbuf->len>0) {
			db_vals[0].val.blob_val.s   = sbuf->s;
			db_vals[0].val.blob_val.len = sbuf->len;
		} else if(t->uas.response.buffer.s==NULL) {
			db_vals[0].val.blob_val.s = "No reply buffer";
			db_vals[0].val.blob_val.len = sizeof("No reply buffer")-1;
		} else {
			db_vals[0].val.blob_val.s = t->uas.response.buffer.s;
			db_vals[0].val.blob_val.len = t->uas.response.buffer.len;
		}
	}
	
	/* check Call-ID header */
	if(msg->callid==NULL || msg->callid->body.s==NULL)
	{
		LM_ERR("cannot find Call-ID header!\n");
		goto error;
	}

	db_keys[1] = callid_column;
	db_vals[1].type = DB_STR;
	db_vals[1].nul = 0;
	db_vals[1].val.str_val.s = msg->callid->body.s;
	db_vals[1].val.str_val.len = msg->callid->body.len;
	
	db_keys[2] = method_column;
//.........这里部分代码省略.........

void query_xcap_update(unsigned int ticks, void* param)
{
	db_key_t query_cols[3], update_cols[3];
	db_val_t query_vals[3], update_vals[3];
	db_key_t result_cols[7];
	int n_result_cols = 0, n_query_cols= 0, n_update_cols= 0;
	db_res_t* result= NULL;
	int user_col, domain_col, doc_type_col, etag_col, doc_uri_col, port_col; 
	db_row_t *row ;	
	db_val_t *row_vals ;
	unsigned int port;
	char* etag, *path, *new_etag= NULL, *doc= NULL;
	int u_doc_col, u_etag_col;
	str user, domain, uri;
	int i;

	/* query the ones I have to handle */
	query_cols[n_query_cols] = "source";
	query_vals[n_query_cols].type = DB_INT;
	query_vals[n_query_cols].nul = 0;
	query_vals[n_query_cols].val.int_val= XCAP_CL_MOD;
	n_query_cols++;

	query_cols[n_query_cols] = "path";
	query_vals[n_query_cols].type = DB_STR;
	query_vals[n_query_cols].nul = 0;

	update_cols[u_doc_col=n_update_cols] = "doc";
	update_vals[n_update_cols].type = DB_STRING;
	update_vals[n_update_cols].nul = 0;
	n_update_cols++;

	update_cols[u_etag_col=n_update_cols] = "etag";
	update_vals[n_update_cols].type = DB_STRING;
	update_vals[n_update_cols].nul = 0;
	n_update_cols++;

	result_cols[user_col= n_result_cols++]     = "username";
	result_cols[domain_col=n_result_cols++]    = "domain";
	result_cols[doc_type_col=n_result_cols++]  = "doc_type";
	result_cols[etag_col=n_result_cols++]      = "etag";
	result_cols[doc_uri_col= n_result_cols++]  = "doc_uri";
	result_cols[port_col= n_result_cols++]     = "port";
	
	if (xcap_dbf.use_table(xcap_db, xcap_db_table) < 0) 
	{
		LM_ERR("in use_table-[table]= %s\n", xcap_db_table);
		goto error;
	}

	if(xcap_dbf.query(xcap_db, query_cols, 0, query_vals, result_cols, 1,
				n_result_cols, 0, &result)< 0)
	{
		LM_ERR("in sql query\n");
		goto error;
	}
	if(result== NULL)
	{
		LM_ERR("in sql query- null result\n");
		return;
	}
	if(result->n<= 0)
	{
		xcap_dbf.free_result(xcap_db, result);
		return;
	}
	n_query_cols++;
	
	/* ask if updated */
	for(i= 0; i< result->n; i++)
	{
		row = &result->rows[i];
		row_vals = ROW_VALUES(row);
	
		path= (char*)row_vals[doc_uri_col].val.string_val;
		port= row_vals[port_col].val.int_val;
		etag= (char*)row_vals[etag_col].val.string_val;	

		user.s= (char*)row_vals[user_col].val.string_val;
		user.len= strlen(user.s);

		domain.s= (char*)row_vals[domain_col].val.string_val;
		domain.len= strlen(domain.s);

		/* send HTTP request */
		doc= send_http_get(path, port, etag, IF_NONE_MATCH, &new_etag);
		if(doc== NULL)
		{
			LM_DBG("document not update\n");
			continue;
		}
		if(new_etag== NULL)
		{
			LM_ERR("etag not found\n");
			pkg_free(doc);
			goto error;
		}
		/* update in xcap db table */
		update_vals[u_doc_col].val.string_val= doc;
		update_vals[u_etag_col].val.string_val= etag;
//.........这里部分代码省略.........