rework presence cleanup

kamailio/db_queries_kazoo.cfg

@@ -12,12 +12,12 @@
 #!substdef "!KZQ_REPLACE_WATCHERS_LOG!REPLACE INTO active_watchers_log (presentity_uri, watcher_username, watcher_domain, event, callid, to_user, to_domain, user_agent, time, result, sent_msg, received_msg) VALUES (\"\$subs(uri)\", \"\$subs(watcher_username)\", \"\$subs(watcher_domain)\", \"\$subs(event)\",\"\$subs(callid)\",\"\$subs(to_user)\",\"\$subs(to_domain)\", '\$(subs(user_agent){s.escape.common}{s.replace,\\\',''}{s.replace,\$\$,})', \$TS, \$notify_reply(\$rs), '\$(mb{s.escape.common}{s.replace,\\\',''}{s.replace,\$\$,})', '\$(notify_reply(\$mb){s.escape.common}{s.replace,\\\',''}{s.replace,\$\$,})')!g"
 
 # # #!substdef "!KZQ_RESET_PUBLISHER_UPDATE!update active_watchers set expires = \$TS where id in (select b.id from presentity a inner join active_watchers b on a.username = b.to_user and a.domain = b.to_domain and a.event = b.event where a.sender = \"\$var(MediaUrl)\")!g"
-#!substdef "!KZQ_RESET_PUBLISHER_UPDATE!INSERT INTO tmp_probe select distinct a.event, a.presentity_uri from presentities a inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where sender = \"\$var(MediaUrl)\" and state in('early', 'confirmed', 'onthephone', 'busy')!g"
+#!substdef "!KZQ_RESET_PUBLISHER_UPDATE!INSERT INTO tmp_probe select distinct a.event, a.presentity_uri, 0 from presentities a inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where sender = \"\$var(MediaUrl)\" and state in('early', 'confirmed', 'onthephone', 'busy')!g"
 #!substdef "!KZQ_PRESENCE_RESET!delete from presentity where sender = \"\$var(MediaUrl)\"!g"
 
 # # #!substdef "!KZQ_RESET_ACCOUNT_UPDATE!update active_watchers set expires = \$TS where watcher_domain=\"\$(kzE{kz.json,Realm})\"!g"
-#!substdef "!KZQ_RESET_ACCOUNT_UPDATE!INSERT INTO tmp_probe select distinct a.event, a.presentity_uri from presentities a inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where domain=\"\$(kzE{kz.json,Realm})\" and state in('early', 'confirmed', 'onthephone', 'busy')!g"
+#!substdef "!KZQ_RESET_ACCOUNT_UPDATE!INSERT INTO tmp_probe select distinct a.event, a.presentity_uri, 0 from presentities a inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where domain=\"\$(kzE{kz.json,Realm})\" and state in('early', 'confirmed', 'onthephone', 'busy')!g"
 #!substdef "!KZQ_RESET_ACCOUNT_RESET!delete from presentity where domain=\"\$(kzE{kz.json,Realm})\"!g"
 
-#!substdef "!KZQ_RESET_PUBLISHER_ZONE_UPDATE!INSERT INTO tmp_probe select distinct a.event, a.presentity_uri from presentities a inner join wdispatcher c on a.sender = c.destination inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where zone = \"\$var(Zone)\" and state in('early', 'confirmed', 'onthephone', 'busy')!g"
+#!substdef "!KZQ_RESET_PUBLISHER_ZONE_UPDATE!INSERT INTO tmp_probe select distinct a.event, a.presentity_uri, 0 from presentities a inner join wdispatcher c on a.sender = c.destination inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where zone = \"\$var(Zone)\" and state in('early', 'confirmed', 'onthephone', 'busy')!g"
 #!substdef "!KZQ_PRESENCE_ZONE_RESET!delete from presentity where id in(select a.id from presentities a join wdispatcher c on a.sender = c.destination where zone = \"\$var(Zone)\")!g"

kamailio/db_scripts/db_extra_check.sql

@@ -17,7 +17,7 @@ KazooDB -db ${DB_CURRENT_DB} "delete from presentity where id in(select id from
 ## notify watchers of pending calls
 ## 'create temp table as' because it will be dropped as soon as we ended the session
 KazooDB -db ${DB_CURRENT_DB} "drop table if exists tmp_probe;"
-KazooDB -db ${DB_CURRENT_DB} "create table tmp_probe as select distinct a.event, a.presentity_uri from presentities a inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where state in('early', 'confirmed', 'onthephone', 'busy');"
+KazooDB -db ${DB_CURRENT_DB} "create table tmp_probe as select distinct a.event, a.presentity_uri, cast(2 as integer) action from presentities a inner join active_watchers b on a.presentity_uri = b.presentity_uri and a.event = b.event where state in('early', 'confirmed', 'onthephone', 'busy');"
 KazooDB -db ${DB_CURRENT_DB} "delete from presentity where id in(select id from presentities where state in('early', 'confirmed', 'onthephone', 'busy'));"
 
 }

kamailio/default.cfg

@@ -971,7 +971,7 @@ event_route[evrexec:DEFERRED_INIT]
     xlog("L_INFO", "processing deferred init\n");
 
     #!ifdef PRESENCE_ROLE
-    route(PRESENCE_PUBLISHER_CLEANUP);
+    route(PRESENCE_DEFERRED_INIT);
     #!endif
 
     #!import_file "custom-init.cfg"

kamailio/presence-notify.cfg

@@ -44,6 +44,9 @@ modparam("rtimer", "timer", "name=notifytimer;interval=500000u;mode=1;")
 modparam("rtimer", "exec", "timer=notifytimer;route=PRESENCE_LOG_TIMER_ROUTE")
 modparam("rtimer", "timer", "name=pres_cleanup;interval=10;mode=1;")
 modparam("rtimer", "exec", "timer=pres_cleanup;route=PRESENCE_CLEANUP")
+modparam("rtimer", "timer", "name=pres_publisher_cleanup;interval=5;mode=1;")
+modparam("rtimer", "exec", "timer=pres_publisher_cleanup;route=PRESENCE_PUBLISHER_CLEANUP")
+
 
 event_route[presence:notify-reply]
 {
@@ -141,22 +144,49 @@ route[PRESENCE_CLEANUP]
 
 route[PRESENCE_PUBLISHER_CLEANUP]
 {
-    xlog("L_INFO", "processing presence publisher cleanup\n");
-    if (sql_xquery("cb", "select * from tmp_probe", "ra") == 1)
-    {
-        while($xavp(ra) != $null) {
-           xlog("L_DEBUG", "processing $xavp(ra=>event) notifies for $xavp(ra=>presentity_uri)\n");
-           pres_refresh_watchers("$xavp(ra=>presentity_uri)", "$xavp(ra=>event)", 1);
-           pv_unset("$xavp(ra)");
+    xlog("L_DEBUG", "processing presence publisher cleanup\n");
+
+    $var(sqlres) = sql_query("cb", "update tmp_probe set action = 1 where action = 0");
+    if($var(sqlres) < 0) {
+        xlog("L_ERROR", "$var(ci)|log|error cleaning tmp_probe\n");
+        return;
+    } else {
+        $var(nrows) = $sqlrows(cb);
+        if($var(nrows) > 0) {
+            if (sql_xquery("cb", "select * from tmp_probe where action = 1", "cleanup_pres") == 1)
+            {
+                while($xavp(cleanup_pres) != $null) {
+                   xlog("L_DEBUG", "processing $xavp(cleanup_pres=>event) notifies for $xavp(cleanup_pres=>presentity_uri)\n");
+                   pres_refresh_watchers("$xavp(cleanup_pres=>presentity_uri)", "$xavp(cleanup_pres=>event)", 1);
+                   pv_unset("$xavp(cleanup_pres)");
+                }
+            }
+
+            $var(sqlres) = sql_query("cb", "delete from tmp_probe where action = 1");
+            if($var(sqlres) < 0) {
+                xlog("L_ERROR", "$var(ci)|log|error cleaning tmp_probe\n");
+            } else {
+                $var(nrows) = $sqlrows(cb);
+                if($var(nrows) > 0) {
+                    xlog("L_DEBUG", "presence publisher cleanup processed $var(nrows) rows\n");
+                }
+            }
         }
     }
-    $var(sqlres) = sql_query("cb", "delete from tmp_probe");
+
+}
+
+route[PRESENCE_DEFERRED_INIT]
+{
+    xlog("L_INFO", "processing presence deferred init\n");
+    $var(sqlres) = sql_query("cb", "update tmp_probe set action = 0 where action = 2");
     if($var(sqlres) < 0) {
         xlog("L_ERROR", "$var(ci)|log|error cleaning tmp_probe\n");
+        return;
     } else {
         $var(nrows) = $sqlrows(cb);
         if($var(nrows) > 0) {
-            xlog("L_INFO", "presence publisher cleanup processed $var(nrows) rows\n");
+           xlog("L_NOTICE", "scheduled update for $var(nrows) watched presentities/event\n");
         }
     }
-}
+}