presence(k): option to fetch records from db in chunks

- can cope with large number of db records
- new parameter fetch_rows (default value 500)

modules_k/presence/presence.c

@@ -90,6 +90,7 @@ str presentity_table= str_init("presentity");
 str active_watchers_table = str_init("active_watchers");
 str watchers_table= str_init("watchers");
 
+int pres_fetch_rows = 500;
 int library_mode= 0;
 str server_address= {0, 0};
 evlist_t* EvList= NULL;
@@ -179,6 +180,7 @@ static param_export_t params[]={
 	{ "enable_sphere_check",    INT_PARAM, &sphere_enable},
 	{ "timeout_rm_subs",        INT_PARAM, &timeout_rm_subs},
 	{ "send_fast_notify",	    INT_PARAM, &send_fast_notify},
+	{ "fetch_rows",             INT_PARAM, &pres_fetch_rows},
     {0,0,0}
 };
 

modules_k/presence/presentity.c

@@ -54,6 +54,8 @@ xmlNodePtr xmlNodeGetNodeByName(xmlNodePtr node, const char *name,
 static str pu_200_rpl  = str_init("OK");
 static str pu_412_rpl  = str_init("Conditional request failed");
 
+extern int pres_fetch_rows;
+
 #define ETAG_LEN  128
 
 char* generate_ETag(int publ_count)
@@ -717,8 +719,8 @@ int pres_htable_restore(void)
 		goto error;
 	}
 	static str query_str = str_init("username");
-	if (pa_dbf.query (pa_db, 0, 0, 0,result_cols,0, n_result_cols,
-				&query_str, &result) < 0)
+	if (db_fetch_query(&pa_dbf, pres_fetch_rows, pa_db, 0, 0, 0, result_cols,
+				0, n_result_cols, &query_str, &result) < 0)
 	{
 		LM_ERR("querying presentity\n");
 		goto error;
@@ -731,58 +733,62 @@ int pres_htable_restore(void)
 		pa_dbf.free_result(pa_db, result);
 		return 0;
 	}
-		
-	for(i= 0; i< result->n; i++)
-	{
-		row = &result->rows[i];
-		row_vals = ROW_VALUES(row);
 
-		if(row_vals[expires_col].val.int_val< (int)time(NULL))
-			continue;
-		
-		sphere= NULL;
-		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);
-		ev_str.s= (char*)row_vals[event_col].val.string_val;
-		ev_str.len= strlen(ev_str.s);
-
-		if(event_parser(ev_str.s, ev_str.len, &ev)< 0)
+	do {
+		for(i= 0; i< result->n; i++)
 		{
-			LM_ERR("parsing event\n");
+			row = &result->rows[i];
+			row_vals = ROW_VALUES(row);
+
+			if(row_vals[expires_col].val.int_val< (int)time(NULL))
+				continue;
+		
+			sphere= NULL;
+			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);
+			ev_str.s= (char*)row_vals[event_col].val.string_val;
+			ev_str.len= strlen(ev_str.s);
+
+			if(event_parser(ev_str.s, ev_str.len, &ev)< 0)
+			{
+				LM_ERR("parsing event\n");
+				free_event_params(ev.params.list, PKG_MEM_TYPE);
+				goto error;
+			}
+			event= ev.type;
 			free_event_params(ev.params.list, PKG_MEM_TYPE);
-			goto error;
-		}
-		event= ev.type;
-		free_event_params(ev.params.list, PKG_MEM_TYPE);
 
-		if(uandd_to_uri(user, domain, &uri)< 0)
-		{
-			LM_ERR("constructing uri\n");
-			goto error;
-		}
-		/* insert in hash_table*/
+			if(uandd_to_uri(user, domain, &uri)< 0)
+			{
+				LM_ERR("constructing uri\n");
+				goto error;
+			}
+			/* insert in hash_table*/
 	
-		if(sphere_enable && event== EVENT_PRESENCE )
-		{
-			body.s= (char*)row_vals[body_col].val.string_val;
-			body.len= strlen(body.s);
-			sphere= extract_sphere(body);
-		}
+			if(sphere_enable && event== EVENT_PRESENCE )
+			{
+				body.s= (char*)row_vals[body_col].val.string_val;
+				body.len= strlen(body.s);
+				sphere= extract_sphere(body);
+			}
 
-		if(insert_phtable(&uri, event, sphere)< 0)
-		{
-			LM_ERR("inserting record in presentity hash table");
-			pkg_free(uri.s);
+			if(insert_phtable(&uri, event, sphere)< 0)
+			{
+				LM_ERR("inserting record in presentity hash table");
+				pkg_free(uri.s);
+				if(sphere)
+					pkg_free(sphere);
+				goto error;
+			}
 			if(sphere)
 				pkg_free(sphere);
-			goto error;
+			pkg_free(uri.s);
 		}
-		if(sphere)
-			pkg_free(sphere);
-		pkg_free(uri.s);
-	}
+	} while((db_fetch_next(&pa_dbf, pres_fetch_rows, pa_db, &result)==1)
+			&& (RES_ROW_N(result)>0));
+
 	pa_dbf.free_result(pa_db, result);
 
 	return 0;

modules_k/presence/subscribe.c

@@ -46,8 +46,6 @@
 #include "notify.h"
 #include "../pua/hash.h"
 
-#define ACTW_FETCH_SIZE  128
-
 int get_stored_info(struct sip_msg* msg, subs_t* subs, int* error_ret,
 		str* reply_str);
 int get_database_info(struct sip_msg* msg, subs_t* subs, int* error_ret,
@@ -61,6 +59,7 @@ static str pu_400_rpl  = str_init("Bad request");
 static str pu_500_rpl  = str_init("Server Internal Error");
 static str pu_489_rpl  = str_init("Bad Event");
 
+extern int pres_fetch_rows;
 
 int send_2XX_reply(struct sip_msg * msg, int reply_code, int lexpire,
 		str* local_contact)
@@ -1837,27 +1836,11 @@ int restore_db_subs(void)
 	}
 
 	/* select the whole table and all the columns */
-	if (DB_CAPABILITY(pa_dbf, DB_CAP_FETCH)) 
+	if (db_fetch_query(&pa_dbf, pres_fetch_rows, pa_db, 0, 0, 0, result_cols,
+				0, n_result_cols, 0, &result) < 0)
 	{
-		if(pa_dbf.query(pa_db,0,0,0,result_cols, 0,
-		n_result_cols, 0, 0) < 0) 
-		{
-			LM_ERR("Error while querying (fetch) database\n");
-			return -1;
-		}
-		if(pa_dbf.fetch_result(pa_db,&result,ACTW_FETCH_SIZE)<0)
-		{
-			LM_ERR("fetching rows failed\n");
-			return -1;
-		}
-	} else 
-	{
-		if (pa_dbf.query (pa_db, 0, 0, 0,result_cols,0, n_result_cols,
-					0, &result) < 0)
-		{
-			LM_ERR("querying presentity\n");
-			goto error;
-		}
+		LM_ERR("querying presentity\n");
+		goto error;
 	}
 
 	nr_rows = RES_ROW_N(result);
@@ -1983,19 +1966,8 @@ int restore_db_subs(void)
 			}
 		}
 
-		/* any more data to be fetched ?*/
-		if (DB_CAPABILITY(pa_dbf, DB_CAP_FETCH)) {
-		    if (pa_dbf.fetch_result( pa_db, &result,
-					     ACTW_FETCH_SIZE ) < 0) {
-			LM_ERR("fetching more rows failed\n");
-			goto error;
-		    }
-		    nr_rows = RES_ROW_N(result);
-		} else {
-		    nr_rows = 0;
-		}
-
-	}while (nr_rows>0);
+	} while((db_fetch_next(&pa_dbf, pres_fetch_rows, pa_db, &result)==1)
+			&& (RES_ROW_N(result)>0));
 
 	pa_dbf.free_result(pa_db, result);