Browse code

modules_k/presence: added new functionality to the presence engine

The new "dbmode" parameter controls the way subscriptions are handled: in memory, fallback to database, or database only.
The "fallback2db" parameter will still ve available for config compatibility.

Marius Bucur authored on 12/01/2011 16:13:32
Showing 13 changed files
... ...
@@ -47,10 +47,11 @@ Juha Heinanen
47 47
               3.9. max_expires (int)
48 48
               3.10. server_address (str)
49 49
               3.11. fallback2db (int)
50
-              3.12. subs_htable_size (int)
51
-              3.13. pres_htable_size (int)
52
-              3.14. enable_sphere_check (int)
53
-              3.15. timeout_rm_subs (int)
50
+              3.12. dbmode (int)
51
+              3.13. subs_htable_size (int)
52
+              3.14. pres_htable_size (int)
53
+              3.15. enable_sphere_check (int)
54
+              3.16. timeout_rm_subs (int)
54 55
 
55 56
         4. Exported Functions
56 57
 
... ...
@@ -98,15 +99,16 @@ Juha Heinanen
98 98
    1.9. Set max_expires parameter
99 99
    1.10. Set server_address parameter
100 100
    1.11. Set fallback2db parameter
101
-   1.12. Set subs_htable_size parameter
102
-   1.13. Set pres_htable_size parameter
103
-   1.14. Set enable_sphere_check parameter
104
-   1.15. Set timeout_rm_subs parameter
105
-   1.16. handle_publish usage
106
-   1.17. handle_subscribe usage
107
-   1.18. pres_auth_status usage
108
-   1.19. pres_refresh_watchers usage
109
-   1.20. pres_update_watchers usage
101
+   1.12. Set dbmode parameter
102
+   1.13. Set subs_htable_size parameter
103
+   1.14. Set pres_htable_size parameter
104
+   1.15. Set enable_sphere_check parameter
105
+   1.16. Set timeout_rm_subs parameter
106
+   1.17. handle_publish usage
107
+   1.18. handle_subscribe usage
108
+   1.19. pres_auth_status usage
109
+   1.20. pres_refresh_watchers usage
110
+   1.21. pres_update_watchers usage
110 111
    2.1. presence_api_t structure
111 112
 
112 113
 Chapter 1. Admin Guide
... ...
@@ -132,10 +134,11 @@ Chapter 1. Admin Guide
132 132
         3.9. max_expires (int)
133 133
         3.10. server_address (str)
134 134
         3.11. fallback2db (int)
135
-        3.12. subs_htable_size (int)
136
-        3.13. pres_htable_size (int)
137
-        3.14. enable_sphere_check (int)
138
-        3.15. timeout_rm_subs (int)
135
+        3.12. dbmode (int)
136
+        3.13. subs_htable_size (int)
137
+        3.14. pres_htable_size (int)
138
+        3.15. enable_sphere_check (int)
139
+        3.16. timeout_rm_subs (int)
139 140
 
140 141
    4. Exported Functions
141 142
 
... ...
@@ -172,7 +175,13 @@ Chapter 1. Admin Guide
172 172
    mode, in case a searched record is not found in cache, the search is
173 173
    continued in database. This is useful for an architecture in which
174 174
    processing and memory load might be divided on several Kamailio
175
-   instances, maybe on different servers using the same database.
175
+   instances, maybe on different servers using the same database. This
176
+   parameter remains only for legacy purposes. As a new feature for the
177
+   presence engine, it is possible to have three database modes, which one
178
+   can configure through the dbmode parameter. Setting dbmode to 0, 1, 2
179
+   respective will cause the subscribers to be retrieved from memory only,
180
+   from memory and to fallback to database mode in case a record is not
181
+   found in memory, and from database only.
176 182
 
177 183
    The module implements several API functions, that can be used by other
178 184
    modules. In fact, it can be used only as a resource module, or
... ...
@@ -211,10 +220,11 @@ Chapter 1. Admin Guide
211 211
    3.9. max_expires (int)
212 212
    3.10. server_address (str)
213 213
    3.11. fallback2db (int)
214
-   3.12. subs_htable_size (int)
215
-   3.13. pres_htable_size (int)
216
-   3.14. enable_sphere_check (int)
217
-   3.15. timeout_rm_subs (int)
214
+   3.12. dbmode (int)
215
+   3.13. subs_htable_size (int)
216
+   3.14. pres_htable_size (int)
217
+   3.15. enable_sphere_check (int)
218
+   3.16. timeout_rm_subs (int)
218 219
 
219 220
 3.1. db_url(str)
220 221
 
... ...
@@ -352,7 +362,21 @@ modparam("presence", "server_address", "sip:10.10.10.10:5060")
352 352
 modparam("presence", "fallback2db", 1)
353 353
 ...
354 354
 
355
-3.12. subs_htable_size (int)
355
+3.12. dbmode (int)
356
+
357
+   This parameter sets the mode in which records are retrieved. Setting
358
+   this parameter to 0 or 1 is equivalent to setting fallback2db to 0 or
359
+   1, respectiv. The dbmode parameter can also take a third value, 2, in
360
+   which records are retrieved from database only. So, the three database
361
+   modes in which the presence engine can operate are: memory only,
362
+   fallback to database, and database only.
363
+
364
+   Example 1.12. Set dbmode parameter
365
+...
366
+modparam("presence", "dbmode", 2)
367
+...
368
+
369
+3.13. subs_htable_size (int)
356 370
 
357 371
    The size of the in-memory hash table to store subscription dialogs.
358 372
    This parameter will be used as the power of 2 when computing table
... ...
@@ -360,24 +384,24 @@ modparam("presence", "fallback2db", 1)
360 360
 
361 361
    Default value is “9 (512)”.
362 362
 
363
-   Example 1.12. Set subs_htable_size parameter
363
+   Example 1.13. Set subs_htable_size parameter
364 364
 ...
365 365
 modparam("presence", "subs_htable_size", 11)
366 366
 ...
367 367
 
368
-3.13. pres_htable_size (int)
368
+3.14. pres_htable_size (int)
369 369
 
370 370
    The size of the in-memory hash table to store publish records. This
371 371
    parameter will be used as the power of 2 when computing table size.
372 372
 
373 373
    Default value is “9 (512)”.
374 374
 
375
-   Example 1.13. Set pres_htable_size parameter
375
+   Example 1.14. Set pres_htable_size parameter
376 376
 ...
377 377
 modparam("presence", "pres_htable_size", 11)
378 378
 ...
379 379
 
380
-3.14. enable_sphere_check (int)
380
+3.15. enable_sphere_check (int)
381 381
 
382 382
    This parameter is a flag that should be set if permission rules include
383 383
    sphere checking. The sphere information is expected to be present in
... ...
@@ -387,12 +411,12 @@ modparam("presence", "pres_htable_size", 11)
387 387
 
388 388
    Default value is “0 ”.
389 389
 
390
-   Example 1.14. Set enable_sphere_check parameter
390
+   Example 1.15. Set enable_sphere_check parameter
391 391
 ...
392 392
 modparam("presence", "enable_sphere_check", 1)
393 393
 ...
394 394
 
395
-3.15. timeout_rm_subs (int)
395
+3.16. timeout_rm_subs (int)
396 396
 
397 397
    This parameter is a flag that should be set if subscriptions should be
398 398
    removed from the active_watchers when a NOTIFY times out. RFC3265
... ...
@@ -402,7 +426,7 @@ modparam("presence", "enable_sphere_check", 1)
402 402
 
403 403
    Default value is “1”.
404 404
 
405
-   Example 1.15. Set timeout_rm_subs parameter
405
+   Example 1.16. Set timeout_rm_subs parameter
406 406
 ...
407 407
 modparam("presence", "timeout_rm_subs", 0)
408 408
 ...
... ...
@@ -435,7 +459,7 @@ modparam("presence", "timeout_rm_subs", 0)
435 435
 
436 436
    The module sends an appropriate stateless reply in all cases.
437 437
 
438
-   Example 1.16. handle_publish usage
438
+   Example 1.17. handle_publish usage
439 439
 ...
440 440
         if(is_method("PUBLISH"))
441 441
         {
... ...
@@ -461,7 +485,7 @@ modparam("presence", "timeout_rm_subs", 0)
461 461
 
462 462
    The module sends an appropriate stateless reply in all cases.
463 463
 
464
-   Example 1.17. handle_subscribe usage
464
+   Example 1.18. handle_subscribe usage
465 465
 ...
466 466
 if(method=="SUBSCRIBE")
467 467
     handle_subscribe();
... ...
@@ -478,7 +502,7 @@ if(method=="SUBSCRIBE")
478 478
 
479 479
    This function can be used from REQUEST_ROUTE.
480 480
 
481
-   Example 1.18. pres_auth_status usage
481
+   Example 1.19. pres_auth_status usage
482 482
 ...
483 483
 if (method=="MESSAGE") {
484 484
     pres_auth_status("$fu", $ru");
... ...
@@ -509,7 +533,7 @@ if (method=="MESSAGE") {
509 509
 
510 510
    This function can be used from ANY_ROUTE.
511 511
 
512
-   Example 1.19. pres_refresh_watchers usage
512
+   Example 1.20. pres_refresh_watchers usage
513 513
 ...
514 514
 pres_refresh_watchers("sip:test@kamailio.org", "presence", 1);
515 515
 ...
... ...
@@ -527,7 +551,7 @@ pres_refresh_watchers("sip:test@kamailio.org", "presence", 1);
527 527
 
528 528
    This function can be used from ANY_ROUTE.
529 529
 
530
-   Example 1.20. pres_update_watchers usage
530
+   Example 1.21. pres_update_watchers usage
531 531
 ...
532 532
 pres_update_watchers("sip:test@kamailio.org", "presence");
533 533
 ...
... ...
@@ -32,6 +32,10 @@
32 32
 	found in cache, the search is continued	in database. This is useful for
33 33
 	an architecture in which processing and memory load might be divided on 
34 34
 	several &kamailio; instances, maybe on different servers using the same database.
35
+        This parameter remains only for legacy purposes. As a new feature for the presence engine, it is possible
36
+        to have three database modes, which one can configure through the dbmode parameter.
37
+        Setting dbmode to 0, 1, 2 respective will cause the subscribers to be retrieved from memory only,
38
+        from memory and to fallback to database mode in case a record is not found in memory, and from database only.
35 39
 	</para>
36 40
 	<para>The module implements several API functions, that can be used by other
37 41
 	modules. In fact, it can be used only as a resource module, or "library".
... ...
@@ -294,7 +298,23 @@ modparam("presence", "fallback2db", 1)
294 294
 </programlisting>
295 295
 		</example>
296 296
 	</section>
297
-
297
+<section>
298
+		<title><varname>dbmode</varname> (int)</title>
299
+		<para>
300
+		This parameter sets the mode in which records are retrieved.
301
+                Setting this parameter to 0 or 1 is equivalent to setting fallback2db to 0 or 1, respectiv.
302
+                The dbmode parameter can also take a third value, 2, in which records are retrieved from database only.
303
+                So, the three database modes in which the presence engine can operate are: memory only, fallback to database, and database only.
304
+		</para>
305
+		<example>
306
+		<title>Set <varname>dbmode</varname> parameter</title>
307
+		<programlisting format="linespecific">
308
+...
309
+modparam("presence", "dbmode", 2)
310
+...
311
+</programlisting>
312
+		</example>
313
+	</section>
298 314
 	<section>
299 315
 		<title><varname>subs_htable_size</varname> (int)</title>
300 316
 		<para>
... ...
@@ -261,8 +261,8 @@ int insert_shtable(shtable_t htable,unsigned int hash_code, subs_t* subs)
261 261
 	}
262 262
 
263 263
 	new_rec->expires+= (int)time(NULL);
264
-	if(fallback2db!=0) {
265
-		if(new_rec->db_flag==0)
264
+	if(dbmode == DB_FALLBACK) {
265
+		if(new_rec->db_flag == 0)
266 266
 			new_rec->db_flag = INSERTDB_FLAG;
267 267
 	} else {
268 268
 		new_rec->db_flag = NO_UPDATEDB_FLAG;
... ...
@@ -380,7 +380,7 @@ str* get_wi_notify_body(subs_t* subs, subs_t* watcher_subs)
380 380
 		goto done;
381 381
 	}
382 382
 
383
-	if(fallback2db)
383
+	if(dbmode != DB_MEMORY_ONLY)
384 384
 	{
385 385
 		if(get_wi_subs_db(subs, watchers)< 0)
386 386
 		{
... ...
@@ -405,7 +405,7 @@ str* get_wi_notify_body(subs_t* subs, subs_t* watcher_subs)
405 405
 			continue;
406 406
 		}
407 407
 
408
-		if(fallback2db && s->db_flag!= INSERTDB_FLAG)
408
+		if(dbmode != DB_MEMORY_ONLY && s->db_flag!= INSERTDB_FLAG)
409 409
 		{
410 410
 			LM_DBG("record already found in database\n");
411 411
 			continue;
... ...
@@ -641,7 +641,7 @@ str* get_p_notify_body(str pres_uri, pres_ev_t* event, str* etag,
641 641
 	if(search_phtable(&pres_uri, event->evp->type, hash_code)== NULL)
642 642
 	{
643 643
 		LM_DBG("No record exists in hash_table\n");
644
-		if(fallback2db)
644
+		if(dbmode != DB_MEMORY_ONLY)
645 645
 			goto db_query;
646 646
 
647 647
 		/* for pidf manipulation */
... ...
@@ -1226,10 +1226,12 @@ subs_t* get_subs_dialog(str* pres_uri, pres_ev_t* event, str* sender)
1226 1226
 	subs_t* s_array= NULL;
1227 1227
 	int n= 0, i= 0;
1228 1228
 	
1229
-	/* if fallback2db -> should take all dialogs from db
1230
-	 * and the only those dialogs from cache with db_flag= INSERTDB_FLAG */
1229
+	/* if in memory mode, should take the subscriptions from the hashtable only
1230
+	   in dbonly mode should take all dialogs from db
1231
+	   in fallback mode, should take those dialogs with db_flag = INSERTDB_FLAG
1232
+	*/
1231 1233
 
1232
-	if(fallback2db)
1234
+	if(dbmode != DB_MEMORY_ONLY)
1233 1235
 	{
1234 1236
 		if(get_subs_db(pres_uri, event, sender, &s_array, &n)< 0)			
1235 1237
 		{
... ...
@@ -1237,68 +1239,73 @@ subs_t* get_subs_dialog(str* pres_uri, pres_ev_t* event, str* sender)
1237 1237
 			goto error;
1238 1238
 		}
1239 1239
 	}
1240
-	hash_code= core_hash(pres_uri, &event->name, shtable_size);
1241 1240
 	
1242
-	lock_get(&subs_htable[hash_code].lock);
1243
-
1244
-	s= subs_htable[hash_code].entries;
1245
-
1246
-	while(s->next)
1241
+	if(dbmode != DB_ONLY)
1247 1242
 	{
1248
-		s= s->next;
1249
-	
1250
-		printf_subs(s);
1243
+		hash_code= core_hash(pres_uri, &event->name, shtable_size);
1251 1244
 		
1252
-		if(s->expires< (int)time(NULL))
1253
-		{
1254
-			LM_DBG("expired subs\n");
1255
-			continue;
1256
-		}
1257
-		
1258
-		if((!(s->status== ACTIVE_STATUS &&
1259
-            s->reason.len== 0 &&
1260
-			s->event== event && s->pres_uri.len== pres_uri->len &&
1261
-			strncmp(s->pres_uri.s, pres_uri->s, pres_uri->len)== 0)) || 
1262
-			(sender && sender->len== s->contact.len && 
1263
-			strncmp(sender->s, s->contact.s, sender->len)== 0))
1264
-			continue;
1245
+		lock_get(&subs_htable[hash_code].lock);
1265 1246
 
1266
-		if(fallback2db)
1247
+		s= subs_htable[hash_code].entries;
1248
+
1249
+		while(s->next)
1267 1250
 		{
1268
-			if(s->db_flag== NO_UPDATEDB_FLAG)
1251
+			s= s->next;
1252
+		
1253
+			printf_subs(s);
1254
+			
1255
+			if(s->expires< (int)time(NULL))
1269 1256
 			{
1270
-				LM_DBG("s->db_flag==NO_UPDATEDB_FLAG\n");
1257
+				LM_DBG("expired subs\n");
1271 1258
 				continue;
1272 1259
 			}
1273 1260
 			
1274
-			if(s->db_flag== UPDATEDB_FLAG)
1261
+			if((!(s->status== ACTIVE_STATUS &&
1262
+		    s->reason.len== 0 &&
1263
+				s->event== event && s->pres_uri.len== pres_uri->len &&
1264
+				strncmp(s->pres_uri.s, pres_uri->s, pres_uri->len)== 0)) || 
1265
+				(sender && sender->len== s->contact.len && 
1266
+				strncmp(sender->s, s->contact.s, sender->len)== 0))
1267
+				continue;
1268
+
1269
+			if(dbmode == DB_FALLBACK)
1275 1270
 			{
1276
-				LM_DBG("s->db_flag== UPDATEDB_FLAG\n");
1277
-				if(n>0 && update_in_list(s, s_array, i, n)< 0)
1271
+				if(s->db_flag== NO_UPDATEDB_FLAG)
1278 1272
 				{
1279
-					LM_DBG("dialog not found in list fetched from database\n");
1280
-					/* insert record */
1273
+					LM_DBG("s->db_flag==NO_UPDATEDB_FLAG\n");
1274
+					continue;
1275
+				}
1276
+				
1277
+				if(s->db_flag== UPDATEDB_FLAG)
1278
+				{
1279
+					LM_DBG("s->db_flag== UPDATEDB_FLAG\n");
1280
+					if(n>0 && update_in_list(s, s_array, i, n)< 0)
1281
+					{
1282
+						LM_DBG("dialog not found in list fetched from database\n");
1283
+						/* insert record */
1284
+					}
1285
+					else
1286
+						continue;			
1281 1287
 				}
1282
-				else
1283
-					continue;			
1284 1288
 			}
1289
+			
1290
+			LM_DBG("s->db_flag= INSERTDB_FLAG\n");
1291
+			s_new= mem_copy_subs(s, PKG_MEM_TYPE);
1292
+			if(s_new== NULL)
1293
+			{
1294
+				LM_ERR("copying subs_t structure\n");
1295
+				lock_release(&subs_htable[hash_code].lock);
1296
+				goto error;
1297
+			}
1298
+			s_new->expires-= (int)time(NULL);
1299
+			s_new->next= s_array;
1300
+			s_array= s_new;
1301
+			i++;
1285 1302
 		}
1286 1303
 		
1287
-		LM_DBG("s->db_flag= INSERTDB_FLAG\n");
1288
-		s_new= mem_copy_subs(s, PKG_MEM_TYPE);
1289
-		if(s_new== NULL)
1290
-		{
1291
-			LM_ERR("copying subs_t structure\n");
1292
-			lock_release(&subs_htable[hash_code].lock);
1293
-			goto error;
1294
-		}
1295
-		s_new->expires-= (int)time(NULL);
1296
-		s_new->next= s_array;
1297
-		s_array= s_new;
1298
-		i++;
1299
-	}
1300
-
1301 1304
 	lock_release(&subs_htable[hash_code].lock);
1305
+	}
1306
+	
1302 1307
 	LM_DBG("found %d dialogs( %d in database and %d in hash_table)\n",n+i,n,i);
1303 1308
 
1304 1309
 	return s_array;
... ...
@@ -1643,30 +1650,33 @@ int notify(subs_t* subs, subs_t * watcher_subs,str* n_body,int force_null_body)
1643 1643
 		unsigned int hash_code;
1644 1644
 		hash_code= core_hash(&subs->pres_uri, &subs->event->name, shtable_size);
1645 1645
 
1646
-		if(update_shtable(subs_htable, hash_code, subs, LOCAL_TYPE)< 0)
1646
+		/* if subscriptions are held also in memory, update the subscription hashtable */
1647
+		if(dbmode != DB_ONLY)
1647 1648
 		{
1648
-			if(subs->db_flag!= INSERTDB_FLAG && fallback2db)
1649
+			if(update_shtable(subs_htable, hash_code, subs, LOCAL_TYPE) < 0 && dbmode == DB_MEMORY_ONLY)
1649 1650
 			{
1650
-				LM_DBG("record not found in subs htable\n");
1651
-				if(update_subs_db(subs, LOCAL_TYPE)< 0)
1652
-				{
1653
-					LM_ERR("updating subscription in database\n");
1654
-					return -1;
1655
-				}
1651
+				/* subscriptions are held only in memory, and hashtable update failed */
1652
+				LM_ERR("updating subscription record in hash table\n");
1653
+				return -1;
1656 1654
 			}
1657
-			else
1655
+		}
1656
+		/* if dbonly mode, or if fallback2db mode and the subscription was inserted into the database */
1657
+		if((dbmode == DB_ONLY) || (subs->db_flag != INSERTDB_FLAG && dbmode == DB_FALLBACK))
1658
+		{
1659
+			LM_DBG("updating subscription to database\n");
1660
+			if(update_subs_db(subs, LOCAL_TYPE)< 0)
1658 1661
 			{
1659
-				LM_ERR("record not found in subs htable\n");
1662
+				LM_ERR("updating subscription in database\n");
1660 1663
 				return -1;
1661 1664
 			}
1662 1665
 		}
1663 1666
 	}
1664 1667
      
1665
-    if(subs->reason.s && subs->status== ACTIVE_STATUS && 
1666
-        subs->reason.len== 12 && strncmp(subs->reason.s, "polite-block", 12)== 0)
1667
-    {
1668
-        force_null_body = 1;
1669
-    }
1668
+	if(subs->reason.s && subs->status== ACTIVE_STATUS && 
1669
+	subs->reason.len== 12 && strncmp(subs->reason.s, "polite-block", 12)== 0)
1670
+	{
1671
+		force_null_body = 1;
1672
+	}
1670 1673
 
1671 1674
 	if(send_notify_request(subs, watcher_subs, n_body, force_null_body)< 0)
1672 1675
 	{
... ...
@@ -69,6 +69,7 @@
69 69
 #include "../../lib/kmi/mi.h"
70 70
 #include "../../lib/kcore/hash_func.h"
71 71
 #include "../pua/hash.h"
72
+#include "presence.h"
72 73
 #include "publish.h"
73 74
 #include "subscribe.h"
74 75
 #include "event_list.h"
... ...
@@ -136,7 +137,8 @@ int expires_offset = 0;
136 136
 int max_expires= 3600;
137 137
 int shtable_size= 9;
138 138
 shtable_t subs_htable= NULL;
139
-int fallback2db= 0;
139
+int dbmode = 0;
140
+int fallback2db = 0;
140 141
 int sphere_enable= 0;
141 142
 int timeout_rm_subs = 1;
142 143
 
... ...
@@ -175,6 +177,7 @@ static param_export_t params[]={
175 175
 	{ "server_address",         STR_PARAM, &server_address.s},
176 176
 	{ "subs_htable_size",       INT_PARAM, &shtable_size},
177 177
 	{ "pres_htable_size",       INT_PARAM, &phtable_size},
178
+	{ "dbmode",                 INT_PARAM, &dbmode},
178 179
 	{ "fallback2db",            INT_PARAM, &fallback2db},
179 180
 	{ "enable_sphere_check",    INT_PARAM, &sphere_enable},
180 181
 	{ "timeout_rm_subs",        INT_PARAM, &timeout_rm_subs},
... ...
@@ -358,6 +361,12 @@ static int mod_init(void)
358 358
 	if(pa_db)
359 359
 		pa_dbf.close(pa_db);
360 360
 	pa_db = NULL;
361
+	
362
+	/* for legacy, we also keep the fallback2db parameter, but make sure for consistency */
363
+	if(fallback2db)
364
+	{
365
+		dbmode = DB_FALLBACK;
366
+	}
361 367
 
362 368
 	return 0;
363 369
 }
... ...
@@ -44,6 +44,15 @@
44 44
 #include "event_list.h"
45 45
 #include "hash.h"
46 46
 
47
+/* DB modes */
48
+
49
+/* subscriptions are held in memory and periodically updated to db, but retrieved from db only at startup */
50
+#define DB_MEMORY_ONLY 0
51
+/* same as memory_only, but if a subscription is not found, it falls back to db */
52
+#define DB_FALLBACK 1
53
+/* subscriptions are held only in database */
54
+#define DB_ONLY 2
55
+
47 56
 /* TM bind */
48 57
 extern struct tm_binds tmb;
49 58
 
... ...
@@ -66,7 +75,7 @@ extern char *to_tag_pref;
66 66
 extern int expires_offset;
67 67
 extern str server_address;
68 68
 extern int max_expires;
69
-extern int fallback2db;
69
+extern int dbmode;
70 70
 extern int sphere_enable;
71 71
 extern int timeout_rm_subs;
72 72
 extern int shtable_size;
... ...
@@ -896,8 +896,8 @@ char* get_sphere(str* pres_uri)
896 896
 	lock_release(&pres_htable[hash_code].lock);
897 897
 
898 898
 
899
-	/* if record not found and fallback2db query database*/
900
-	if(!fallback2db)
899
+	/* if record not found and subscriptions are held also in database, query database*/
900
+	if(dbmode == DB_MEMORY_ONLY)
901 901
 	{
902 902
 		return NULL;
903 903
 	}
... ...
@@ -180,6 +180,152 @@ int delete_db_subs(str pres_uri, str ev_stored_name, str to_tag)
180 180
 	return 0;
181 181
 }
182 182
 
183
+int insert_subs_db(subs_t* s, int type)
184
+{
185
+	db_key_t query_cols[22];
186
+	db_val_t query_vals[22];
187
+	int n_query_cols = 0;
188
+	int pres_uri_col, to_user_col, to_domain_col, from_user_col, from_domain_col,
189
+		callid_col, totag_col, fromtag_col, event_col,status_col, event_id_col, 
190
+		local_cseq_col, remote_cseq_col, expires_col, record_route_col, 
191
+		contact_col, local_contact_col, version_col,socket_info_col,reason_col;
192
+		
193
+	if(pa_dbf.use_table(pa_db, &active_watchers_table)< 0)
194
+	{
195
+		LM_ERR("sql use table failed\n");
196
+		return -1;
197
+	}
198
+	
199
+	query_cols[pres_uri_col= n_query_cols] =&str_presentity_uri_col;
200
+	query_vals[pres_uri_col].type = DB1_STR;
201
+	query_vals[pres_uri_col].nul = 0;
202
+	n_query_cols++;
203
+	
204
+	query_cols[callid_col= n_query_cols] =&str_callid_col;
205
+	query_vals[callid_col].type = DB1_STR;
206
+	query_vals[callid_col].nul = 0;
207
+	n_query_cols++;
208
+
209
+	query_cols[totag_col= n_query_cols] =&str_to_tag_col;
210
+	query_vals[totag_col].type = DB1_STR;
211
+	query_vals[totag_col].nul = 0;
212
+	n_query_cols++;
213
+
214
+	query_cols[fromtag_col= n_query_cols] =&str_from_tag_col;
215
+	query_vals[fromtag_col].type = DB1_STR;
216
+	query_vals[fromtag_col].nul = 0;
217
+	n_query_cols++;
218
+
219
+	query_cols[to_user_col= n_query_cols] =&str_to_user_col;
220
+	query_vals[to_user_col].type = DB1_STR;
221
+	query_vals[to_user_col].nul = 0;
222
+	n_query_cols++;
223
+
224
+	query_cols[to_domain_col= n_query_cols] =&str_to_domain_col;
225
+	query_vals[to_domain_col].type = DB1_STR;
226
+	query_vals[to_domain_col].nul = 0;
227
+	n_query_cols++;
228
+	
229
+	query_cols[from_user_col= n_query_cols] =&str_watcher_username_col;
230
+	query_vals[from_user_col].type = DB1_STR;
231
+	query_vals[from_user_col].nul = 0;
232
+	n_query_cols++;
233
+
234
+	query_cols[from_domain_col= n_query_cols] =&str_watcher_domain_col;
235
+	query_vals[from_domain_col].type = DB1_STR;
236
+	query_vals[from_domain_col].nul = 0;
237
+	n_query_cols++;
238
+
239
+	query_cols[event_col= n_query_cols] =&str_event_col;
240
+	query_vals[event_col].type = DB1_STR;
241
+	query_vals[event_col].nul = 0;
242
+	n_query_cols++;	
243
+
244
+	query_cols[event_id_col= n_query_cols] =&str_event_id_col;
245
+	query_vals[event_id_col].type = DB1_STR;
246
+	query_vals[event_id_col].nul = 0;
247
+	n_query_cols++;
248
+
249
+	query_cols[local_cseq_col= n_query_cols]=&str_local_cseq_col;
250
+	query_vals[local_cseq_col].type = DB1_INT;
251
+	query_vals[local_cseq_col].nul = 0;
252
+	n_query_cols++;
253
+
254
+	query_cols[remote_cseq_col= n_query_cols]=&str_remote_cseq_col;
255
+	query_vals[remote_cseq_col].type = DB1_INT;
256
+	query_vals[remote_cseq_col].nul = 0;
257
+	n_query_cols++;
258
+
259
+	query_cols[expires_col= n_query_cols] =&str_expires_col;
260
+	query_vals[expires_col].type = DB1_INT;
261
+	query_vals[expires_col].nul = 0;
262
+	n_query_cols++;
263
+
264
+	query_cols[status_col= n_query_cols] =&str_status_col;
265
+	query_vals[status_col].type = DB1_INT;
266
+	query_vals[status_col].nul = 0;
267
+	n_query_cols++;
268
+
269
+	query_cols[reason_col= n_query_cols] =&str_reason_col;
270
+	query_vals[reason_col].type = DB1_STR;
271
+	query_vals[reason_col].nul = 0;
272
+	n_query_cols++;
273
+
274
+	query_cols[record_route_col= n_query_cols] =&str_record_route_col;
275
+	query_vals[record_route_col].type = DB1_STR;
276
+	query_vals[record_route_col].nul = 0;
277
+	n_query_cols++;
278
+	
279
+	query_cols[contact_col= n_query_cols] =&str_contact_col;
280
+	query_vals[contact_col].type = DB1_STR;
281
+	query_vals[contact_col].nul = 0;
282
+	n_query_cols++;
283
+
284
+	query_cols[local_contact_col= n_query_cols] =&str_local_contact_col;
285
+	query_vals[local_contact_col].type = DB1_STR;
286
+	query_vals[local_contact_col].nul = 0;
287
+	n_query_cols++;
288
+
289
+	query_cols[socket_info_col= n_query_cols] =&str_socket_info_col;
290
+	query_vals[socket_info_col].type = DB1_STR;
291
+	query_vals[socket_info_col].nul = 0;
292
+	n_query_cols++;
293
+
294
+	query_cols[version_col= n_query_cols]=&str_version_col;
295
+	query_vals[version_col].type = DB1_INT;
296
+	query_vals[version_col].nul = 0;
297
+	n_query_cols++;
298
+	
299
+	query_vals[pres_uri_col].val.str_val= s->pres_uri;
300
+	query_vals[callid_col].val.str_val= s->callid;
301
+	query_vals[totag_col].val.str_val= s->to_tag;
302
+	query_vals[fromtag_col].val.str_val= s->from_tag;
303
+	query_vals[to_user_col].val.str_val = s->to_user;
304
+	query_vals[to_domain_col].val.str_val = s->to_domain;
305
+	query_vals[from_user_col].val.str_val = s->from_user;
306
+	query_vals[from_domain_col].val.str_val = s->from_domain;
307
+	query_vals[event_col].val.str_val = s->event->name;
308
+	query_vals[event_id_col].val.str_val = s->event_id;
309
+	query_vals[local_cseq_col].val.int_val= s->local_cseq;
310
+	query_vals[remote_cseq_col].val.int_val= s->remote_cseq;
311
+	query_vals[expires_col].val.int_val = s->expires + (int)time(NULL);
312
+	query_vals[record_route_col].val.str_val = s->record_route;
313
+	query_vals[contact_col].val.str_val = s->contact;
314
+	query_vals[local_contact_col].val.str_val = s->local_contact;
315
+	query_vals[version_col].val.int_val= s->version;
316
+	query_vals[status_col].val.int_val= s->status;
317
+	query_vals[reason_col].val.str_val= s->reason;
318
+	query_vals[socket_info_col].val.str_val= s->sockinfo_str;
319
+
320
+	LM_DBG("inserting subscription in active_watchers table\n");
321
+	if(pa_dbf.insert(pa_db, query_cols, query_vals, n_query_cols) < 0)
322
+	{
323
+		LM_ERR("unsuccessful sql insert\n");
324
+		return -1;
325
+	}
326
+	return 0;
327
+}
328
+
183 329
 int update_subs_db(subs_t* subs, int type)
184 330
 {
185 331
 	db_key_t query_cols[22], update_keys[7];
... ...
@@ -316,10 +462,11 @@ int update_subscription(struct sip_msg* msg, subs_t* subs, int to_tag_gen,
316 316
 				LM_ERR("deleting subscription record from database\n");
317 317
 				goto error;
318 318
 			}
319
-			/* delete record from hash table also */
320
-
321
-			subs->local_cseq= delete_shtable(subs_htable,hash_code,
322
-					subs->to_tag);
319
+			/* delete record from hash table also if not in dbonly mode */
320
+			if(dbmode != DB_ONLY)
321
+			{
322
+				subs->local_cseq= delete_shtable(subs_htable, hash_code, subs->to_tag);
323
+			}
323 324
 		
324 325
 			if(subs->event->type & PUBL_TYPE)
325 326
 			{	
... ...
@@ -359,16 +506,21 @@ int update_subscription(struct sip_msg* msg, subs_t* subs, int to_tag_gen,
359 359
 			}
360 360
 			return 1;
361 361
 		}
362
-
363
-		if(update_shtable(subs_htable, hash_code, subs, REMOTE_TYPE)< 0)
362
+		/* if subscribers are held in memory, update them */
363
+		if(dbmode != DB_ONLY)
364 364
 		{
365
-			if(fallback2db==0)
365
+			if(update_shtable(subs_htable, hash_code, subs, REMOTE_TYPE)< 0)
366 366
 			{
367
-				LM_ERR("updating subscription record in hash table\n");
368
-				goto error;
367
+				/* if subscribers are also retrieved from database, it is not a fatal error */
368
+				if(dbmode != DB_MEMORY_ONLY)
369
+				{
370
+					LM_ERR("updating subscription record in hash table\n");
371
+					goto error;
372
+				}
369 373
 			}
370 374
 		}
371
-		if(fallback2db!=0)
375
+		/* if subscribers are retrieved from db also, update the subscription in database immediately */
376
+		if(dbmode != DB_MEMORY_ONLY)
372 377
 		{
373 378
 			/* update in database table */
374 379
 			if(update_subs_db(subs, REMOTE_TYPE)< 0)
... ...
@@ -380,16 +532,30 @@ int update_subscription(struct sip_msg* msg, subs_t* subs, int to_tag_gen,
380 380
 	}
381 381
 	else
382 382
 	{
383
+		LM_DBG("subscription not in dialog\n");
383 384
 		if(subs->expires!= 0)
384
-		{	
385
-			if(insert_shtable(subs_htable,hash_code,subs)< 0)
385
+		{
386
+			if(dbmode != DB_ONLY)
386 387
 			{
387
-				LM_ERR("inserting new record in subs_htable\n");
388
-				goto error;
388
+				LM_DBG("inserting in shtable\n");
389
+				if(insert_shtable(subs_htable,hash_code,subs)< 0)
390
+				{
391
+					LM_ERR("inserting new record in subs_htable\n");
392
+					goto error;
393
+				}
389 394
 			}
395
+			else
396
+			{
397
+				if(insert_subs_db(subs, REMOTE_TYPE));
398
+			}
399
+			/* TODO if req_auth, the subscription was in the watcher table first, we must delete it */
390 400
 		}
391 401
 		/*otherwise there is a subscription outside a dialog with expires= 0 
392 402
 		 * no update in database, but should try to send Notify */
403
+		else
404
+		{
405
+			LM_DBG("subscription request with expiry=0 not in dialog\n");
406
+		}
393 407
 	}
394 408
 
395 409
 /* reply_and_notify  */
... ...
@@ -527,7 +693,7 @@ int handle_subscribe(struct sip_msg* msg, char* str1, char* str2)
527 527
 	reply_code= 500;
528 528
 	reply_str= pu_500_rpl;
529 529
 
530
-	if( parse_headers(msg,HDR_EOH_F, 0)==-1 )
530
+	if(parse_headers(msg,HDR_EOH_F, 0) == -1)
531 531
 	{
532 532
 		LM_ERR("parsing headers\n");
533 533
 		reply_code= 400;
... ...
@@ -602,7 +768,7 @@ int handle_subscribe(struct sip_msg* msg, char* str1, char* str2)
602 602
 			goto error;
603 603
 		}
604 604
 		reason= subs.reason;
605
-	}	
605
+	}
606 606
 
607 607
 	/* call event specific subscription handling */
608 608
 	if(event->evs_subs_handl)
... ...
@@ -653,7 +819,7 @@ int handle_subscribe(struct sip_msg* msg, char* str1, char* str2)
653 653
 					LM_ERR("in event specific function is_watcher_allowed\n");
654 654
 					goto error;
655 655
 				}
656
-				if(get_status_str(subs.status)== NULL)
656
+				if(get_status_str(subs.status) == NULL)
657 657
 				{
658 658
 					LM_ERR("wrong status= %d\n", subs.status);
659 659
 					goto error;
... ...
@@ -678,7 +844,7 @@ int handle_subscribe(struct sip_msg* msg, char* str1, char* str2)
678 678
 		LM_ERR("wrong status\n");
679 679
 		goto error;
680 680
 	}
681
-    LM_DBG("subscription status= %s - %s\n", get_status_str(subs.status), 
681
+	LM_DBG("subscription status= %s - %s\n", get_status_str(subs.status), 
682 682
             found==0?"inserted":"found in watcher table");
683 683
 	
684 684
 	if(update_subscription(msg, &subs, to_tag_gen, &sent_reply) <0)
... ...
@@ -1045,7 +1211,7 @@ int get_stored_info(struct sip_msg* msg, subs_t* subs, int* reply_code,
1045 1045
 		lock_release(&subs_htable[i].lock);
1046 1046
 	}
1047 1047
 
1048
-	if(fallback2db)
1048
+	if(dbmode == DB_FALLBACK)
1049 1049
 	{
1050 1050
 		return get_database_info(msg, subs, reply_code, reply_str);	
1051 1051
 	}
... ...
@@ -1326,7 +1492,7 @@ int handle_expired_subs(subs_t* s)
1326 1326
 void timer_db_update(unsigned int ticks,void *param)
1327 1327
 {	
1328 1328
 	int no_lock=0;
1329
-
1329
+	LM_DBG("db_update timer\n");
1330 1330
 	if(ticks== 0 && param == NULL)
1331 1331
 		no_lock= 1;
1332 1332
 	
... ...
@@ -1498,6 +1664,13 @@ void update_db_subs(db1_con_t *db,db_func_t dbf, shtable_t hash_table,
1498 1498
 		LM_ERR("null database connection\n");
1499 1499
 		return;
1500 1500
 	}
1501
+	
1502
+	/* if in dbonly mode, no update to database is required */
1503
+	if(dbmode == DB_ONLY)
1504
+	{
1505
+		goto delete_expired_subs;
1506
+	}
1507
+	
1501 1508
 	for(i=0; i<htable_size; i++) 
1502 1509
 	{
1503 1510
 		if(!no_lock)
... ...
@@ -1602,6 +1775,7 @@ void update_db_subs(db1_con_t *db,db_func_t dbf, shtable_t hash_table,
1602 1602
 			lock_release(&hash_table[i].lock);	
1603 1603
 	}
1604 1604
 
1605
+delete_expired_subs:
1605 1606
 	update_vals[0].val.int_val= (int)time(NULL) - expires_offset;
1606 1607
 	update_ops[0]= OP_LT;
1607 1608
 	if(dbf.delete(db, update_cols, update_ops, update_vals, 1) < 0)
... ...
@@ -1801,7 +1975,7 @@ int restore_db_subs(void)
1801 1801
 			s.sockinfo_str.s=(char*)row_vals[sockinfo_col].val.string_val;
1802 1802
 			s.sockinfo_str.len= strlen(s.sockinfo_str.s);
1803 1803
 
1804
-			if(fallback2db!=0)
1804
+			if(dbmode == DB_FALLBACK)
1805 1805
 				s.db_flag = NO_UPDATEDB_FLAG;
1806 1806
 			hash_code= core_hash(&s.pres_uri, &s.event->name, shtable_size);
1807 1807
 			if(insert_shtable(subs_htable, hash_code, &s)< 0)
... ...
@@ -1828,7 +2002,7 @@ int restore_db_subs(void)
1828 1828
 	pa_dbf.free_result(pa_db, result);
1829 1829
 
1830 1830
 	/* delete all records */
1831
-	if(fallback2db==0 && pa_dbf.delete(pa_db, 0,0,0,0)< 0)
1831
+	if(dbmode != DB_MEMORY_ONLY && pa_dbf.delete(pa_db, 0,0,0,0)< 0)
1832 1832
 	{
1833 1833
 		LM_ERR("deleting all records from database table\n");
1834 1834
 		return -1;
1835 1835
new file mode 100755
... ...
@@ -0,0 +1,13 @@
0
+#!/bin/bash
1
+killall -9 sipp &> /dev/null
2
+
3
+mysql_username="-u${MYSQL_USERNAME}"
4
+if [[ ${MYSQL_PASSWORD} ]]; then
5
+	mysql_pass="-p$2"
6
+	echo "da"
7
+else
8
+	mysql_pass=""
9
+fi
10
+
11
+mysql ${mysql_username} ${mysql_pass} ${MYSQL_DATABASE} -e 'truncate table watchers'
12
+mysql ${mysql_username} ${mysql_pass} ${MYSQL_DATABASE} -e 'truncate table active_watchers'
0 13
new file mode 100755
... ...
@@ -0,0 +1,12 @@
0
+#!/bin/bash
1
+
2
+export MYSQL_USERNAME="root"
3
+export MYSQL_PASSWORD=""
4
+export MYSQL_DATABASE="openser"
5
+export EXIT_FAILURE=2
6
+export SIPP_SCEN_DIR="../"
7
+export MI_HOST="127.0.0.1"
8
+export KAMAILIO_HOST="127.0.0.1"
9
+export SUBSCRIBE_WAIT_SECONDS=8
10
+export NGREP_LOG_FILE="ngrep.log"
11
+export SUBSCRIBERS_FILE="usernames.txt"
0 12
new file mode 100755
... ...
@@ -0,0 +1,58 @@
0
+#!/bin/bash
1
+
2
+source config.sh
3
+
4
+# some basic checking for dependencies
5
+if [[ `whoami` != "root" ]]; then
6
+	echo "you must be root to run the suite..."
7
+	exit $EXIT_FAILURE;
8
+fi
9
+if [[ ! `which sipp` ]]; then
10
+	echo "you do not have sipp installed..."
11
+	exit $EXIT_FAILURE;
12
+fi
13
+
14
+if [[ ! `which ngrep` ]]; then
15
+	echo "you do not have ngrep installed..."
16
+	exit $EXIT_FAILURE;
17
+fi
18
+
19
+if [[ ! $1 ]]; then
20
+	echo "please choose the number of subscribers..."
21
+	exit $EXIT_FAILURE;
22
+else
23
+	subscribers_no=$1
24
+fi
25
+
26
+echo "starting suite..."
27
+
28
+# truncating log file
29
+cat /dev/null > ${NGREP_LOG_FILE}
30
+# killing sipp
31
+killall -9 sipp &> /dev/null
32
+# running ngrep
33
+ngrep -d any -W byline port 5060 &> ${NGREP_LOG_FILE} &
34
+
35
+./clean_subscribers.sh && ./send_subscribe.sh ${subscribers_no} && ./send_publish.sh
36
+
37
+sleep ${SUBSCRIBE_WAIT_SECONDS}
38
+
39
+# killing ngrep
40
+killall -9 ngrep &> /dev/null
41
+
42
+# get the number of notifies sent
43
+notify_response=`grep 'NOTIFY sip' ${NGREP_LOG_FILE} | wc -l`
44
+
45
+echo received ${notify_response} responses...
46
+
47
+let responses=${subscribers_no}*${subscribers_no}+${subscribers_no}
48
+if [[ $responses == ${notify_response} ]]; then
49
+	echo that look\'s ok...
50
+else
51
+	echo should have received $responses responses...
52
+fi
53
+
54
+# cleaning up the sipp stderr output file
55
+sed s/'Resolving remote host.*'// -i errors.txt
56
+sed s/'Done.$'// -i errors.txt
57
+sed '/^$/d' -i errors.txt
0 58
new file mode 100755
... ...
@@ -0,0 +1,29 @@
0
+#!/bin/bash
1
+
2
+source ../functions/register.sh
3
+source ../functions/subscribe.sh
4
+source ../functions/publish.sh
5
+source ../functions/notify.sh
6
+source ../functions/rand.sh
7
+
8
+EXPECTED_RETURN=0
9
+event="presence"
10
+start_port=8020
11
+content_type="application\/pidf+xml"
12
+usernames_txt="${SUBSCRIBERS_FILE}"
13
+if [[ $1 ]]; then
14
+	usernames_txt=$1
15
+fi
16
+line=`cat $usernames_txt | wc -l`
17
+line_no=`head -1 /dev/urandom | od -N 1 | awk '{ print $2 }'`
18
+let line_no=$line_no%$line
19
+let line_no=$line_no+1
20
+username=`cat $usernames_txt | sed -n ${line_no}p`
21
+for i in `seq 1 $line`; do
22
+	let start_port=$start_port+1
23
+	ADDITIONAL_PARAMETERS="-p $start_port -timeout 20"
24
+	notify > /dev/null 2>> errors.txt
25
+done
26
+sleep 4
27
+publish $event $content_type $username ${KAMAILIO_HOST}
28
+wait
0 29
new file mode 100755
... ...
@@ -0,0 +1,47 @@
0
+#!/bin/bash -e
1
+
2
+source ../functions/register.sh
3
+source ../functions/subscribe.sh
4
+source ../functions/publish.sh
5
+source ../functions/notify.sh
6
+source ../functions/rand.sh
7
+
8
+cat /dev/null > errors.txt
9
+cat /dev/null > ${SUBSCRIBERS_FILE}
10
+
11
+EXPECTED_RETURN=0
12
+start_port=8020
13
+event="presence"
14
+content_type="application\/pidf+xml"
15
+control_port=40400
16
+mi_port=20200
17
+
18
+for i in `seq 1 $1`
19
+do 
20
+	let start_port=$start_port+1
21
+	let mi_port=$mi_port+100
22
+	let control_port=$control_port+100
23
+	usernames[$i]=$i
24
+	#`rand_md5` is more professional
25
+	local_ports[$i]=$start_port
26
+	mi_ports[$i]=$mi_port
27
+	control_ports[$i]=$control_port
28
+done
29
+
30
+k=0
31
+for i in `seq 1 $1`
32
+do
33
+	echo ${usernames[$i]} >> ${SUBSCRIBERS_FILE}
34
+	for j in `seq 1 $1`
35
+	do
36
+		echo subscribing ${usernames[$j]} to ${usernames[$i]} on port ${local_ports[$j]}...
37
+		ADDITIONAL_PARAMETERS=" -mi ${MI_HOST} -mp ${mi_ports[$j]} -cp ${control_ports[$j]} -p ${local_ports[$j]}"
38
+		subscribe $event $content_type ${usernames[$i]} ${KAMAILIO_HOST} 3600 ${usernames[$j]} 2>> errors.txt > /dev/null &
39
+		pids[$k]=$!
40
+		let k=$k+1
41
+	done
42
+	wait
43
+done
44
+
45
+wait
46
+