early handle of subscription timeout

nodes timers, proxy, publish, log - 4.1

kamailio/default.cfg

@@ -201,6 +201,27 @@ modparam("kazoo", "amqp_connection", "MY_AMQP_SECONDARY_URL")
 #!ifdef MY_AMQP_TERTIARY_URL
 modparam("kazoo", "amqp_connection", "MY_AMQP_TERTIARY_URL")
 #!endif
+#!ifdef MY_AMQP_QUATERNARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_QUATERNARY_URL")
+#!endif
+#!ifdef MY_AMQP_QUINARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_QUINARY_URL")
+#!endif
+#!ifdef MY_AMQP_SENARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_SENARY_URL")
+#!endif
+#!ifdef MY_AMQP_SEPTENARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_SEPTENARY_URL")
+#!endif
+#!ifdef MY_AMQP_OCTONARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_OCTONARY_URL")
+#!endif
+#!ifdef MY_AMQP_NONARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_NONARY_URL")
+#!endif
+#!ifdef MY_AMQP_DENARY_URL
+modparam("kazoo", "amqp_connection", "MY_AMQP_DENARY_URL")
+#!endif
 
 ###### kazoo bindings ######
 include_file "kazoo-bindings.cfg"
@@ -356,6 +377,12 @@ route[CHECK_RETRANS]
 
 route[SANITY_CHECK]
 {
+    ## CVE-2018-14767
+    if($(hdr(To)[1]) != $null) {
+        xlog("second To header not null - dropping message");
+        drop;
+    }
+    
     if (!sanity_check()) {
         xlog("L_WARN", "$ci|end|message from $si:$sp is insane\n");
         exit;
@@ -369,6 +396,8 @@ route[SANITY_CHECK]
 
     if ($ua == "friendly-scanner" ||
         $ua == "sundayddr" ||
+        $ua == "pplsip" ||
+        $ua =~ "NiceGuy" ||
         $ua =~ "sipcli" ) {
         xlog("L_WARN", "$ci|end|dropping message with user-agent $ua from $si:$sp\n");
         exit;
@@ -411,6 +440,9 @@ route[HANDLE_OPTIONS]
 
 route[HANDLE_NOTIFY]
 {
+    if (has_totag())
+       return;
+
     if (is_method("NOTIFY")) {
         if (isflagset(FLAG_INTERNALLY_SOURCED)) {
             if (loose_route()) {
@@ -527,6 +559,10 @@ route[HANDLE_IN_DIALOG_REQUESTS]
             if (is_method("INVITE") && !isflagset(FLAG_INTERNALLY_SOURCED) && is_audio_on_hold()) {
                 setflag(FLAG_ASSOCIATE_USER);
             }
+            if ( is_method("NOTIFY") ) {
+                 # Add Record-Route for in-dialog NOTIFY as per RFC 6665.
+                 record_route();
+            }
             route(RELAY);
         } else if (isflagset(FLAG_INTERNALLY_SOURCED)) {
             xlog("L_INFO", "$ci|log|relay internally sourced in-dialog message without loose_route\n");
@@ -835,9 +871,9 @@ route[INTERNAL_REDIRECT]
 
 onsend_route {
     if (isflagset(FLAG_ASSOCIATE_USER)) {
-        $var(contact_uri) = $(ct{tobody.user}) + "@" + $(ct{tobody.host});
-        xlog("L_INFO", "$ci|log|associate user $var(contact_uri) with media server sip:$sndto(ip):$sndto(port)\n");
-        $sht(associations=>$var(contact_uri))= "sip:" + $sndto(ip) + ":" + $sndto(port);
+        $var(user_source) = $(ct{tobody.user}) + "@" + $si + ":" + $sp;
+        xlog("L_INFO", "$ci|log|associate traffic from $var(user_source) with media server sip:$sndto(ip):$sndto(port)\n");
+        $sht(associations=>$var(user_source))= "sip:" + $sndto(ip) + ":" + $sndto(port);
     }
 
     xlog("L_INFO", "$ci|pass|$sndfrom(ip):$sndfrom(port) -> $sndto(ip):$sndto(port)\n");

kamailio/defs.cfg

@@ -83,4 +83,8 @@
 #!endif
 #!endif
 
+#!ifndef PRESENCE_AMQP_PUBLISH
+#!define PRESENCE_AMQP_PUBLISH 1
+#!endif
+
 # vim: tabstop=4 softtabstop=4 shiftwidth=4 expandtab

kamailio/dispatcher-role.cfg

@@ -92,7 +92,7 @@ route[DISPATCHER_FIND_ROUTES]
         }
     }
 
-    $var(contact_uri) = $(ct{tobody.user}) + "@" + $(ct{tobody.host});
+    $var(user_source) = $(ct{tobody.user}) + "@" + $si + ":" + $sp;
     $var(redirect) = @from.uri.user + "@" + @from.uri.host + "->"
         + @ruri.user + "@" + @ruri.host;
     if ($sht(redirects=>$var(redirect)) != $null) {
@@ -101,9 +101,9 @@ route[DISPATCHER_FIND_ROUTES]
         if (route(DISPATCHER_REORDER_ROUTES)) {
             $avp(AVP_REDIRECT_KEY) = $var(redirect);
         }
-    } else if ($sht(associations=>$var(contact_uri)) != $null) {
-        $var(prefered_route) = $sht(associations=>$var(contact_uri));
-        xlog("L_INFO", "$ci|log|found association for contact uri $var(contact_uri)\n");
+    } else if ($sht(associations=>$var(user_source)) != $null) {
+        $var(prefered_route) = $sht(associations=>$var(user_source));
+        xlog("L_INFO", "$ci|log|found association for contact uri $var(user_source)\n");
         if (!route(DISPATCHER_REORDER_ROUTES)) {
             $sht(associations=>$var(association)) = $null;
         }

kamailio/kazoodb.sql

@@ -1,5 +1,6 @@
 PRAGMA foreign_keys=OFF;
 PRAGMA wal=on;
+PRAGMA journal_mode=WAL;
 BEGIN TRANSACTION;
 CREATE TABLE version (
     table_name VARCHAR(32) NOT NULL,

kamailio/nodes-role.cfg

@@ -5,20 +5,23 @@
 #!endif
 
 #!ifndef NODES_FUDGE_FACTOR
-#!define NODES_FUDGE_FACTOR 3
+#!define NODES_FUDGE_FACTOR 10
 #!endif
 
-modparam("htable", "htable", "nodes=>size=8;initval=0;autoexpire=180");
-modparam("htable", "htable", "media=>size=8;initval=0;autoexpire=180");
+modparam("htable", "htable", "nodes=>size=8;initval=0;autoexpire=60");
+modparam("htable", "htable", "media=>size=8;initval=0;autoexpire=60");
 
-####### TIMER module ##########
-#!ifndef TIMER_LOADED
-loadmodule "timer.so"
-#!trydef TIMER_LOADED
+####### RTIMER module ##########
+#!ifndef RTIMER_LOADED
+loadmodule "rtimer.so"
+#!trydef RTIMER_LOADED
 #!endif
-modparam("timer", "declare_timer", "NODES_ADVERTISE_TIMER=NODES_ADVERTISE_ROUTE,5000,slow,enable");
-modparam("timer", "declare_timer", "NODE_TRACK_TIMER=NODE_TRACK_ROUTE,500,fast,enable");
-modparam("timer", "declare_timer", "NODE_HEARTBEAT_TIMER=NODE_HEARTBEAT_ROUTE,500,fast,enable");
+modparam("rtimer", "timer", "name=ta;interval=2;mode=2;")
+modparam("rtimer", "timer", "name=retry;interval=5;mode=2;")
+modparam("rtimer", "timer", "name=pub;interval=10;mode=1;")
+modparam("rtimer", "exec", "timer=ta;route=NODE_HEARTBEAT_ROUTE")
+modparam("rtimer", "exec", "timer=retry;route=NODE_TRACK_ROUTE")
+modparam("rtimer", "exec", "timer=pub;route=NODES_ADVERTISE_ROUTE")
 
 
 modparam("mqueue","mqueue", "name=node_track")
@@ -29,15 +32,11 @@ modparam("mqueue","mqueue", "name=node_heartbeat")
 route[NODES_ADVERTISE_ROUTE]
 {
 
-#!ifdef REGISTRAR_ROLE
-    $var(Registrar) = $_s("Registrar" : {"Registrations" : $(stat(registered_users){s.int})});
-#!else
-    $var(Registrar) = "";
-#!endif
+route(LISTENER_STATUS);
 
 #!ifdef DISPATCHER_ROLE
     route(DISPATCHER_STATUS);
-    $var(Dispatcher) = $_s("Dispatcher" : {"Groups" : { $var(ds_groups_json) }});
+    $var(Dispatcher) = $_s(, "Dispatcher" : {"Groups" : { $var(ds_groups_json) }});
 #!else
     $var(Dispatcher) = "";
 #!endif
@@ -50,13 +49,19 @@ route[NODES_ADVERTISE_ROUTE]
     route(COUNT_ALL_PRESENTITIES);
     $var(Presentities) = $_s("Presentities" : {"message-summary" : $xavp(watchers=>message-summary), "dialog" : $xavp(watchers=>dialog), "presence" : $xavp(watchers=>presence)});
 
-    $var(Presence) = $_s("Presence" : {$var(Subscribers), $var(Subscriptions), $var(Presentities)});
+    $var(Presence) = $_s(, "Presence" : {$var(Subscribers), $var(Subscriptions), $var(Presentities)});
 #!else
     $var(Presence) = "";
 #!endif
 
-    $var(Roles) = $_s("Roles" : {$var(Dispatcher), $var(Presence) , $var(Registrar)});
-    $var(Payload) = '{"Event-Category" : "nodes", "Event-Name" : "advertise", "Expires" : 5000, "Used-Memory" : $(stat(real_used_size){s.int}), "Startup" : $Tb, "XWhApps" : {"kamailio" : {"Startup" : $Tb }},  $var(Roles)}';
+#!ifdef REGISTRAR_ROLE
+    $var(Registrar) = $_s(, "Registrar" : {"Registrations" : $(stat(registered_users){s.int})});
+#!else
+    $var(Registrar) = "";
+#!endif
+
+    $var(Roles) = $_s("Roles" : {"Proxy" : $var(listeners) $var(Dispatcher) $var(Presence) $var(Registrar)});
+    $var(Payload) = '{"Event-Category" : "nodes", "Event-Name" : "advertise", "Expires" : 15000, "Used-Memory" : $(stat(real_used_size){s.int}), "Startup" : $Tb, "WhApps" : {"kamailio" : {"Startup" : $Tb }},  $var(Roles)}';
     kazoo_publish("nodes", "", $var(Payload));
 }
 
@@ -65,11 +70,13 @@ event_route[kazoo:consumer-event-nodes-advertise]
 {
     $var(count) = $shtinc(nodes=>$(kzE{kz.json,Node})::count); 
     if($var(count) == 0) {
-       xlog("L_INFO", "$(kzE{kz.json,Msg-ID})|nodes|hearbeat for reconnected node $(kzE{kz.json,Node})\n");
+       xlog("L_WARNING", "$(kzE{kz.json,Msg-ID})|nodes|hearbeat for reconnected node $(kzE{kz.json,Node})\n");
        $var(count) = $shtinc(nodes=>$(kzE{kz.json,Node})::count);
     } else {
        if($var(count) == 1) {
-          xlog("L_INFO", "$(kzE{kz.json,Msg-ID})|nodes|hearbeat from new node $(kzE{kz.json,Node})\n");
+          xlog("L_WARNING", "$(kzE{kz.json,Msg-ID})|nodes|hearbeat from new node $(kzE{kz.json,Node})\n");
+       } else {
+          xlog("L_DEBUG", "$(kzE{kz.json,Msg-ID})|nodes|hearbeat from existing node $(kzE{kz.json,Node})\n");
        }
     }   
     mq_add("node_heartbeat", "$(kzE{kz.json,Node})", "$kzE");
@@ -79,13 +86,13 @@ event_route[htable:expired:nodes]
 {
     if($shtrecord(key) =~ "::count$$") {
        if($shtrecord(value) == -1) {
-          xlog("L_INFO", "htable|nodes|node $(shtrecord(key){s.rm,::count}) is still unreachable\n");
+          xlog("L_WARNING", "htable|nodes|node $(shtrecord(key){s.rm,::count}) is still unreachable\n");
        }
        mq_add("node_track", "$shtrecord(key)", "");
        return;
     }
 
-    xlog("L_INFO", "htable|nodes|hearbeat expired for node $shtrecord(key)\n");
+    xlog("L_WARNING", "htable|nodes|hearbeat expired for node $shtrecord(key)\n");
 }
 
 route[NODE_TRACK_ROUTE]
@@ -105,6 +112,7 @@ route[NODE_HEARTBEAT_ROUTE]
   while(mq_fetch("node_heartbeat") == 1 && $var(runloop) < MAX_WHILE_LOOPS) {
      $var(Node) = $mqk(node_heartbeat);
      $var(Payload) = $mqv(node_heartbeat);
+     xlog("L_DEBUG", "$(var(Payload){kz.json,Msg-ID})|nodes|processing hearbeat for node $var(Node)\n");
 
      route(CHECK_MEDIA_SERVERS);
      
@@ -130,21 +138,24 @@ route[CHECK_MEDIA_SERVERS]
        $var(Count) = $cnt($avp(MediaKeys));
        $var(Idx) = 0;
        while( $var(Idx) < $var(Count) ) {
-         $var(MediaKey) = $(avp(MediaKeys)[$var(Idx)]{s.replace,$var(Dot),$var(Perc)});
+         $var(MediaName) = $(avp(MediaKeys)[$var(Idx)]);
+         $var(MediaKey) = $(var(MediaName){s.replace,$var(Dot),$var(Perc)});
          avp_delete("$avp(ProfileKeys)/g");
          if(kazoo_json_keys($var(Payload), "Media-Servers.$var(MediaKey).Interfaces", "$avp(ProfileKeys)") == 1) {
             $var(ProfileCount) = $cnt($avp(ProfileKeys));
             $var(ProfileIdx) = 0;
             while( $var(ProfileIdx) < $var(ProfileCount) ) {
-                 $var(MediaRawUrl) = $(var(Payload){kz.json,Media-Servers.$var(MediaKey).Interfaces.$(avp(ProfileKeys)[$var(ProfileIdx)]).URL});
+                 $var(MediaProfile) = $(avp(ProfileKeys)[$var(ProfileIdx)]);
+                 $var(MediaRawUrl) = $(var(Payload){kz.json,Media-Servers.$var(MediaKey).Interfaces.$var(MediaProfile).URL});
                  $var(MediaUrl) =  $(var(MediaRawUrl){re.subst,/^sip:(.*)@(.*)/sip:\2/});
                  if($shtinc(media=>$var(MediaUrl)::count) == 1) {
                      $sht(media=>$var(MediaUrl)::zone) = $var(Zone);
                      $shtex(media=>$var(MediaUrl)::zone) = 0;
                      route(MEDIA_SERVER_UP);
                  };
-                 $shtex(media=>$var(MediaUrl)::count) = ($(var(Payload){kz.json,Expires}{s.int}) / 1000) + NODES_FUDGE_FACTOR;
-                 
+                 $var(MediaExpire) = ($(var(Payload){kz.json,Expires}{s.int}) / 1000) + NODES_FUDGE_FACTOR;
+                 xlog("L_DEBUG", "nodes|media|$var(Node) media expiration $var(MediaExpire) for $var(MediaUrl)\n");              
+                 $shtex(media=>$var(MediaUrl)::count) = $var(MediaExpire);
                  $var(ProfileIdx) = $var(ProfileIdx) + 1;
             }
          }
@@ -153,7 +164,6 @@ route[CHECK_MEDIA_SERVERS]
    };
 }
 
-
 event_route[htable:expired:media]
 {
     $var(MediaUrl) = $(shtrecord(key){re.subst,/(.*)::(.*)/\1/});
@@ -163,7 +173,7 @@ event_route[htable:expired:media]
 
 route[MEDIA_SERVER_UP]
 {
-   xlog("L_INFO", "nodes|media|$var(Node) reported new media server $var(MediaUrl) in zone $var(Zone)\n");
+   xlog("L_WARNING", "nodes|media|$var(Node) reported new media server $var(MediaUrl) in zone $var(Zone)\n");
 
 #!ifdef DISPATCHER_ROLE
    route(DISPATCHER_CHECK_MEDIA_SERVER);
@@ -177,7 +187,7 @@ route[MEDIA_SERVER_UP]
 
 route[MEDIA_SERVER_DOWN]
 {
-   xlog("L_INFO", "htable|media|hearbeat expired for media server $var(MediaUrl) in zone $var(Zone)\n");
+   xlog("L_WARNING", "htable|media|hearbeat expired for media server $var(MediaUrl) in zone $var(Zone)\n");
    
 #!ifdef PRESENCE_ROLE
    route(RESET_PUBLISHER);
@@ -193,6 +203,31 @@ route[NODES_BINDINGS]
 }
 #!endif
 
+route[LISTENER_STATUS]
+{
+    jsonrpc_exec('{"jsonrpc": "2.0", "method": "corex.list_sockets", "id": 1}');
+    $var(count) = $(jsonrpl(body){kz.json.count,result});
+    $var(loop) = 0;
+    $var(sep) = "";
+    $var(listeners) = "";
+    while( $var(loop) < $var(count) ) {
+       $var(listener) = $(jsonrpl(body){kz.json,result[$var(loop)]});
+       $var(proto) = $(var(listener){kz.json,PROTO});
+       $var(address) = $(var(listener){kz.json,ADDRLIST.ADDR});
+       $var(port) = $(var(listener){kz.json,PORT});
+       $var(uri) = $_s($var(proto):$var(address):$var(port));
+       if($(var(listener){kz.json,ADVERTISE}) != "-") {
+          $var(advertise) = $_s( , "advertise" : "$(var(listener){kz.json,ADVERTISE})");
+       } else {
+          $var(advertise) = "";
+       }
+       $var(x) = $_s("$var(uri)" : {"proto" : "$var(proto)", "address" : "$var(address)", "port" : $var(port) $var(advertise) }); 
+       $var(listeners) = $_s($var(listeners)$var(sep)$var(x));
+       $var(loop) = $var(loop) + 1;
+       $var(sep) = " , ";
+    }
+    $var(listeners) = $_s({"Listeners" : { $var(listeners) }});
+}
 
 
 # vim: tabstop=4 softtabstop=4 shiftwidth=4 expandtab

kamailio/presence-role.cfg

@@ -4,7 +4,7 @@
 #!trydef PRESENCE_MAX_EXPIRES 3600
 
 modparam("htable", "htable", "p=>size=32;autoexpire=3600;")
-modparam("htable", "htable", "first=>size=32;autoexpire=3600;initval =0;updateexpire=1")
+modparam("htable", "htable", "first=>size=32;autoexpire=3600;initval =0;updateexpire=1;")
 
 loadmodule "presence.so"
 loadmodule "presence_dialoginfo.so"
@@ -53,6 +53,7 @@ modparam("nat_traversal", "keepalive_state_file", "KAZOO_DATA_DIR/keep_alive_sta
 modparam("nat_traversal", "keepalive_interval", 45)
 #!endif
 
+kazoo.presence_sync_amqp = PRESENCE_AMQP_PUBLISH descr "sync subscriptions to amqp"
 
 #!ifdef FAST_PICKUP_ROLE
 #!include_file "fast-pickup-role.cfg"
@@ -69,13 +70,12 @@ modparam("nat_traversal", "keepalive_interval", 45)
 #!ifdef NAT_TRAVERSAL_ROLE
 route[PRESENCE_NAT]
 {
+    force_rport();
     if (client_nat_test("3")) {
-        fix_contact();
+        if(is_first_hop())
+           set_contact_alias();
     }
-
     nat_keepalive();
-    force_rport();
-
 }
 #!endif
 
@@ -89,18 +89,31 @@ route[HANDLE_SUBSCRIBE]
       route(PRESENCE_NAT);
     #!endif
 
+    if(has_totag()) {
+        loose_route();
+    }
+
+    record_route();
+
+    if (!t_newtran()) {
+        sl_reply_error();
+        exit;
+    }
+    
     if(has_totag()) {
         route(HANDLE_RESUBSCRIBE);
     } else {
         route(HANDLE_NEW_SUBSCRIBE);
     }
 
+    t_release();
+
     exit;
 }
 
 route[HANDLE_RESUBSCRIBE]
 {
-    loose_route();
+        
     if(handle_subscribe()) {
        if($subs(remote_cseq) < 5) {
           $sht(first=>$subs(callid)) = $null;
@@ -158,9 +171,11 @@ route[HANDLE_NEW_SUBSCRIBE]
 
 route[SUBSCRIBE_AMQP]
 {
-    $var(rk) = "subscribe." + $(subs(to_domain){kz.encode}) + "." + $(subs(to_user){kz.encode});
-    $var(amqp_payload_request) = $_s({"Event-Category" : "presence", "Event-Name" : "subscription", "Event-Package" : "$hdr(event)", "Expires" : $subs(expires), "Queue" : "BLF-MY_HOSTNAME", "Server-ID" : "BLF-MY_HOSTNAME" , "Contact" : "$(ct{s.escape.common}{s.replace,\','}{s.replace,$$,})", "Call-ID" : "$ci", "From" : "$fu", "User" : "$subs(uri)", "User-Agent" : "$(ua{s.escape.common}{s.replace,\','}{s.replace,$$,})" });
-    kazoo_publish("omnipresence", "$var(rk)", $var(amqp_payload_request));
+    if(@cfg_get.kazoo.presence_sync_amqp == 1) {
+        $var(rk) = "subscribe." + $(subs(to_domain){kz.encode}) + "." + $(subs(to_user){kz.encode});
+        $var(amqp_payload_request) = $_s({"Event-Category" : "presence", "Event-Name" : "subscription", "Event-Package" : "$hdr(event)", "Expires" : $subs(expires), "Queue" : "BLF-MY_HOSTNAME", "Server-ID" : "BLF-MY_HOSTNAME" , "Contact" : "$(ct{s.escape.common}{s.replace,\','}{s.replace,$$,})", "Call-ID" : "$ci", "From" : "$fu", "User" : "$subs(uri)", "User-Agent" : "$(ua{s.escape.common}{s.replace,\','}{s.replace,$$,})" });
+        kazoo_publish("omnipresence", "$var(rk)", $var(amqp_payload_request));
+    }
 }
 
 route[HANDLE_PUBLISH]
@@ -182,7 +197,7 @@ route[HANDLE_PUBLISH]
 route[COUNT_PRESENTITIES]
 {
     $var(Query) = $_s(select event, (select count(*) from presentity b where username = "$(kzE{kz.json,From}{uri.user})" and domain = "$(kzE{kz.json,From}{uri.domain})" and b.event = a.event) count from event_list a);
-    $var(p) = "uri="+$(kzE{kz.json,From});
+    $var(p) = $_s(uri=$var(presentity));
     if (sql_xquery("cb", "$var(Query)", "subs") == 1)
     {
         while($xavp(subs) != $null) {
@@ -247,10 +262,11 @@ event_route[kazoo:consumer-event-presence-dialog-update]
    route(FAST_PICKUP_INIT);
    #!endif
 
+   $var(presentity) = $(kzE{kz.json,From});
+
    if($(kzE{kz.json,State}) == "terminated") {
        route(COUNT_PRESENTITIES);
    } else {
-       $var(presentity) = $(kzE{kz.json,From});
        route(COUNT_SUBSCRIBERS);
    }
    
@@ -291,15 +307,30 @@ event_route[kazoo:consumer-event-presence-mwi-update]
 
 event_route[kazoo:consumer-event-presence-update]
 {
-   xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|received presence update for $(kzE{kz.json,From})\n");
-   $var(presentity) = $(kzE{kz.json,From});
-   route(COUNT_SUBSCRIBERS);
+   xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|received presence update for $(kzE{kz.json,Presence-ID}) : $kzE\n");
+
+   $var(presentity) = $_s(sip:$(kzE{kz.json,Presence-ID}));
+   
+   if($(kzE{kz.json,State}) == "terminated") {
+       route(COUNT_PRESENTITIES);
+   } else {
+       route(COUNT_SUBSCRIBERS);
+   }
+   
+   if($xavp(watchers=>dialog) > 0) {
+      xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|publishing $var(presentity) dialog update for $xavp(watchers=>dialog) watchers\n");
+      kazoo_pua_publish_dialoginfo($kzE);
+      pres_refresh_watchers("$var(presentity)", "dialog", 1);
+   } else {
+      xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|skip dialog update for $var(presentity)\n");
+   }
+   
    if($xavp(watchers=>presence) > 0) {
-      xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|publishing $(kzE{kz.json,From}) presence update for $xavp(watchers=>presence) watchers\n");
+      xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|publishing $var(presentity) presence update for $xavp(watchers=>presence) watchers\n");
       kazoo_pua_publish_presence($kzE);
-      pres_refresh_watchers("$(kzE{kz.json,From})", "presence", 1);
+      pres_refresh_watchers("$var(presentity)", "presence", 1);
   } else {
-      xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|skip presence update for $(kzE{kz.json,From})\n");
+      xlog("L_INFO", "$(kzE{kz.json,Call-ID})|log|skip presence update for $var(presentity)\n");
   }
 
 }
@@ -317,7 +348,7 @@ route[PRESENCE_BINDINGS]
    $var(payload) = "{ 'exchange' : 'presence', 'type' : 'topic', 'queue' : 'presence-presence-MY_HOSTNAME', 'routing' : 'update.*.*', 'exclusive' : 0, 'federate' : 1 }";
    kazoo_subscribe("$var(payload)");
 
-   $var(payload) = "{ 'exchange' : 'presence', 'type' : 'topic', 'queue' : 'presence-mwi-MY_HOSTNAME', 'routing' : 'mwi_updates.*', 'exclusive' : 0 , 'federate' : 1 }";
+   $var(payload) = "{ 'exchange' : 'presence', 'type' : 'topic', 'queue' : 'presence-mwi-MY_HOSTNAME', 'routing' : 'mwi_updates.*.*', 'exclusive' : 0 , 'federate' : 1 }";
    kazoo_subscribe("$var(payload)");
    #!endif
 

kamailio/presence_notify_sync-role.cfg

@@ -4,7 +4,7 @@ kazoo.presence_notify_log_body = 0 descr "logs the body sent in the notification
 kazoo.presence_notify_log_resp_body = 0 descr "logs the body received from notification"
 kazoo.presence_notify_log_to_table = 1 descr "logs notify/reply to active_watchers_log table"
 kazoo.presence_notify_log_to_amqp = 0 descr "logs notify/reply to amqp"
-kazoo.presence_notify_record_route = 0 descr "add record route header to notify msg sent"
+kazoo.presence_notify_record_route = 1 descr "add record route header to notify msg sent"
 kazoo.presence_notify_log_init_body = 0 descr "logs the body before its sent"
 kazoo.presence_notify_force_send_socket = 0 descr "forces the send socket to the contact"
 
@@ -35,11 +35,13 @@ route[PRESENCE_LOCAL_NOTIFY]
 
 modparam("mqueue","mqueue", "name=presence_last_notity")
 
-#!ifndef TIMER_LOADED
-loadmodule "timer.so"
-#!trydef TIMER_LOADED
+####### RTIMER module ##########
+#!ifndef RTIMER_LOADED
+loadmodule "rtimer.so"
+#!trydef RTIMER_LOADED
 #!endif
-modparam("timer", "declare_timer", "PRESENCE_LOG_TIMER=PRESENCE_LOG_TIMER_ROUTE,500,fast,enable");
+modparam("rtimer", "timer", "name=notifytimer;interval=500000u;mode=2;")
+modparam("rtimer", "exec", "timer=notifytimer;route=PRESENCE_LOG_TIMER_ROUTE")
 
 
 event_route[presence:notify-reply]
@@ -49,11 +51,13 @@ event_route[presence:notify-reply]
 
     $xavp(pres=>delete_subscription) = 0;
 
-    if($notify_reply($rs) == 200) {
+    if($subs(reason) == "timeout") {
+       $xavp(pres=>delete_subscription) = 1;
+       xlog("L_INFO", "$ci|end|deleting subscribtion $subs(pres_uri) for $subs(watcher_username)@$subs(watcher_domain) due to timeout\n");
+    } else if($notify_reply($rs) == 200) {
        $sht(notify=>$ci) = $null;
+       $sht(notify=>$ci::count) = 0;
        xlog("L_INFO", "$ci|end|notified $subs(watcher_username)@$subs(watcher_domain) on behalf of $subs(pres_uri)\n");
-    } else if($notify_reply($rs) == 481 && $subs(reason) == "timeout") {
-       xlog("L_INFO","$ci|end|sent subscription $hdr(Subscription-State)\n");
     } else {
        xlog("L_ERROR", "$ci|error|received $notify_reply($rs) when notifying $subs(watcher_username)@$subs(watcher_domain) on behalf of $subs(pres_uri)\n");
        if($rP != "UDP") {
@@ -88,9 +92,9 @@ route[PRESENCE_LOG_TIMER_ROUTE]
   $var(runloop) = 1;
   while(mq_fetch("presence_last_notity") == 1 && $var(runloop) < MAX_WHILE_LOOPS) {
      $var(ci) = $mqk(presence_last_notity);
-     xlog("L_DEBUG", "Query : $mqv(presence_last_notity)");
+     xlog("L_DEBUG", "Query : $mqv(presence_last_notity)\n");
      $var(sqlres) = sql_query("cb", "$mqv(presence_last_notity)");
-     xlog("L_DEBUG", "Query result : $var(sqlres)");
+     xlog("L_DEBUG", "Query result : $var(sqlres)\n");
      if($var(sqlres) < 0) {
         xlog("L_ERROR", "$var(ci)|log|error updating active_watchers_log\n");
      } else {

kamailio/presence_query-role.cfg

@@ -47,7 +47,7 @@ route[PRESENCE_SEARCH_SUMMARY]
     }
     xlog("L_DEBUG", "$ci| RESULT \"Subscriptions\" : { $var(Subs) }\n");
 
-    $var(amqp_payload_request) = '{"Event-Category" : "presence", "Event-Name" : "search_resp", "Msg-ID" : "$(kzE{kz.json,Msg-ID})", "Subscriptions" : { $var(Subs) } }';
+    $var(amqp_payload_request) = $_s({"Event-Category" : "presence", "Event-Name" : "search_resp", "Msg-ID" : "$(kzE{kz.json,Msg-ID})", "Subscriptions" : { $var(Subs) } });
     kazoo_publish("targeted", "$var(Queue)", $var(amqp_payload_request));
 
 }

kamailio/registrar-role.cfg

@@ -29,7 +29,7 @@ modparam("usrloc", "db_check_update", 1)
 modparam("usrloc", "timer_interval", 30)
 modparam("usrloc", "timer_procs", 1)
 modparam("usrloc", "db_timer_clean", 1)
-
+modparam("usrloc", "fetch_rows", 400)
 
 ######## NAT Traversal module - signaling functions ########
 #!ifdef NAT_TRAVERSAL_ROLE
@@ -53,7 +53,7 @@ modparam("registrar", "regid_mode", 1)
 modparam("registrar", "path_mode", 1)
 modparam("registrar", "use_path", 1)
 modparam("registrar", "received_param", "")
-modparam("registrar", "xavp_rcd", "ulrcd")
+##modparam("registrar", "xavp_rcd", "ulrcd")
 
 ####### Registrar Logic ########
 route[HANDLE_REGISTER]
@@ -281,11 +281,11 @@ route[SAVE_LOCATION]
        $var(ip) = "[" + $Ri + "]";
     }
 
-    $var(amqp_payload_request) = $_s({"Event-Category" : "directory", "Event-Name" : "reg_success", "Status" : "$var(Status)", "Event-Timestamp" : $TS, "Expires" : $(var(expires){s.int}), "First-Registration" : $var(new_reg), "Contact" : "$(ct{s.escape.common}{s.replace,\','}{s.replace,$$,})", "Call-ID" : "$ci", "Realm" : "$fd", "Username" : "$fU", "From-User" : "$fU", "From-Host" : "$fd", "To-User" : "$tU", "To-Host" : "$td", "User-Agent" : "$(ua{s.escape.common}{s.replace,\','}{s.replace,$$,})" , "Custom-Channel-Vars" : $xavp(ulattrs=>custom_channel_vars), "Proxy-Path" : "sip:$var(ip)", "Proxy-IP" : "$var(ip)", "Proxy-Port" : "$Rp", "RUID" : "$xavp(ulrcd=>ruid)", "Source-IP": "$si", "Source-Port": "$sp" });
+    $var(amqp_payload_request) = $_s({"Event-Category" : "directory", "Event-Name" : "reg_success", "Status" : "$var(Status)", "Event-Timestamp" : $TS, "Expires" : $(var(expires){s.int}), "First-Registration" : $var(new_reg), "Contact" : "$(ct{s.escape.common}{s.replace,\','}{s.replace,$$,})", "Call-ID" : "$ci", "Realm" : "$fd", "Username" : "$fU", "From-User" : "$fU", "From-Host" : "$fd", "To-User" : "$tU", "To-Host" : "$td", "User-Agent" : "$(ua{s.escape.common}{s.replace,\','}{s.replace,$$,})" , "Custom-Channel-Vars" : $xavp(ulattrs=>custom_channel_vars), "Proxy-Path" : "sip:$var(ip)", "Proxy-IP" : "$var(ip)", "Proxy-Port" : "$Rp", "Source-IP": "$si", "Source-Port": "$sp" });
     $var(amqp_routing_key) = "registration.success." + $(fd{kz.encode}) + "." + $(fU{kz.encode});
     kazoo_publish("registrar", $var(amqp_routing_key), $var(amqp_payload_request));
 
-    xlog("L_INFO", "$ci|end|successful $(var(Status){s.tolower}) with contact $xavp(ulrcd=>ruid) : $ct\n");
+    xlog("L_INFO", "$ci|end|successful $(var(Status){s.tolower}) with contact : $ct\n");
 
     #!ifdef PUSHER_ROLE
     route(PUSHER_ON_REGISTRATION);