Merge pull request #67 from 2600hz/HELP-36142

handle pusher failover in kamailio

kamailio/default.cfg

@@ -657,22 +657,16 @@ route[INTERNAL_TO_EXTERNAL_RELAY]
                 xlog("L_INFO", "$ci|end|routing to contact $ru\n");
             } else {
                 xlog("L_INFO", "$ci|end|lookup for AOR $hdr(X-KAZOO-AOR) failed\n");
-                sl_send_reply("404", "Not registered");
+                sl_send_reply("410", "Not registered");
                 exit;
             }
         } else if (reg_fetch_contacts("location", "$hdr(X-KAZOO-AOR)", "callee")) {
             $du = $(ulc(callee=>received));
             $fs = $(ulc(callee=>socket));
             xlog("L_INFO", "$ci|log|routing $hdr(X-KAZOO-AOR) to $du via $fs\n");
-        } else if ($hdr(X-KAZOO-PUSHER-Token-ID) != $null) {
-            xlog("L_INFO", "$ci|log|ignoring missing registration while waiting for push notification response\n");
-            t_on_reply("EXTERNAL_REPLY");
-            t_set_fr(0, 10000);
-            t_release();
-            exit;
         } else {
             xlog("L_INFO", "$ci|end|user is not registered\n");
-            sl_send_reply("404", "Not registered");
+            sl_send_reply("410", "Not registered");
             exit;
         }
     }

kamailio/pusher-role.cfg

@@ -1,50 +1,147 @@
 ## PUSHER ROLE
 
-####### SQL OPS module ##########
-#!ifndef TSILO_LOADED
-loadmodule "tsilo.so"
-modparam("tsilo", "use_domain", 1)
-#!trydef TSILO_LOADED
-#!endif
+######## Generic Hash Table container in shared memory ########
+modparam("htable", "htable", "push_cache=>autoexpire=60;")
 
 route[PUSHER_ROUTE]
 {
     if ( (!is_method("INVITE")) || (!isflagset(FLAG_INTERNALLY_SOURCED)) || $hdr(X-KAZOO-PUSHER-Token-ID) == $null)
        return;
 
-    if(registered("location") && $hdr(X-KAZOO-PUSHER-Invite-Format) != "push_and_invite") {
-       return;
+    xlog("L_INFO", "$ci|pusher|start deliver call to $hdr(X-KAZOO-PUSHER-Token-ID)\n");
+    if(route(PUSHER_PREPARE_PUSH)) {
+        if(lookup("location", "$hdr(X-KAZOO-AOR)") > 0) {
+           send_reply(100, "calling a push device");
+           route(PUSHER_TO_EXTERNAL_RELAY);
+        } else {
+           send_reply(100, "waking the push device");
+           route(PUSHER_SEND_PUSH_NOTIFICATION);
+        }
+    }
+    exit();
+}
+
+route[PUSHER_TO_EXTERNAL_RELAY]
+{
+    remove_hf_re("^X-.*");
+    t_set_fr(0, 2000);
+    t_relay();
+}
+
+failure_route[PUSHER_EXTERNAL_FAULT]
+{
+    xlog("L_INFO", "$ci|pusher|fault reply $T_reply_code\n");
+    if (!t_check_status("486|603") && $avp(push_sent) != 1) {
+       send_reply(182, "waking the dead guy");
+       route(PUSHER_SEND_PUSH_NOTIFICATION);
+    } else {
+       t_reply("$T_reply_code", "Pusher Failed");
     }
+}
 
-    xlog("L_INFO", "$ci| pusher received request to wakeup $tu\n");
-    sl_send_reply(180, "waking the dead guy");
+route[PUSHER_PREPARE_PUSH]
+{
     if (t_newtran()) {
-          route(SEND_PUSH_NOTIFICATION);
+          t_set_fr(20000, 20000);
+          route(PUSHER_PREPARE_PUSH_PAYLOAD);
           remove_hf_re("^X-.*");
+          t_on_reply("EXTERNAL_REPLY");
+          t_on_failure("PUSHER_EXTERNAL_FAULT");
           t_save_lumps();
-          ts_store("$(tu{s.tolower})");
-       } else {
+          t_set_auto_inv_100(0);
+          return 1;
+    } else {
           sl_send_reply(500, "error creating transaction for waking the dead guy");
+          return 0;
     }
 }
 
-route[SEND_PUSH_NOTIFICATION]
+route[PUSHER_PREPARE_PUSH_PAYLOAD]
 {
+     
     $var(TokenID) = $hdr(X-KAZOO-PUSHER-Token-ID);
     $var(TokenType) = $hdr(X-KAZOO-PUSHER-Token-Type);
     $var(TokenApp) = $hdr(X-KAZOO-PUSHER-Token-App);
+    $var(TokenProxy) = $hdr(X-KAZOO-PUSHER-Token-Proxy);
+
+    ### token for fast reg ###
+    $var(TokenReg) = $uuid(g);
+    $sht(push_cache=>$var(TokenReg)) = 1;
+
+    ### caller-id ###
     $var(rp) = $hdr(Remote-Party-ID);
-    $var(from) = $(var(rp){tobody.user}) + " - " + $(var(rp){tobody.display}{s.escape.common});
-    $var(Payload) = '{ "Event-Category" : "notification", "Event-Name" : "push_req",  "Call-ID" : "$ci", "Token-ID" : "$var(TokenID)", "Token-Type" : "$var(TokenType)", "Token-App" : "$var(TokenApp)", "Alert-Body" : "$var(from)" }';        
-    $var(RoutingKey) = "notification.push." + $var(TokenType) + "." + $var(TokenID);
-    $var(exchange) = "pushes";
-    kazoo_publish($var(exchange), $var(RoutingKey), $var(Payload));
+    $var(from_user) = $(var(rp){tobody.user});
+    $var(from_name) = $(var(rp){tobody.display}{re.subst,/"//g});
+    $var(from) = $_s($var(from_user) - $var(from_name));
+
+
+    $var(PushPayload) = $_s({"call-id" : "$ci", "proxy" : "$var(TokenProxy)", "caller-id-number" : "$var(from_user)", "caller-id-name" : "$var(from_name)", "registration-token" : "$var(TokenReg)"});
+    $var(Payload) = $_s({ "Event-Category" : "notification", "Event-Name" : "push_req",  "Call-ID" : "$ci", "Token-ID" : "$var(TokenID)", "Token-Type" : "$var(TokenType)", "Token-App" : "$var(TokenApp)", "Alert-Key" : "IC_MSG", "Alert-Params" : ["$var(from)"], "Sound" : "ring.caf", "Payload" : $var(PushPayload) });
+            
+    $avp(push_routing_key) = "notification.push." + $var(TokenType) + "." + $var(TokenID);
+    $avp(push_payload) = $var(Payload);
+}
+
+route[PUSHER_SEND_PUSH_NOTIFICATION]
+{
+    xlog("L_INFO", "$ci|pusher|sending push notification request\n");
+    xlog("L_DEBUG", "$ci|pusher|pushing to $avp(push_routing_key) : $avp(push_payload)\n");
+    t_set_fr(20000, 20000);
+    $avp(push_sent) = 1;
+    t_suspend();
+    $sht(push_cache=>$(tu{s.tolower})) = $_s(a=0;index=$T(id_index);label=$T(id_label));
+    kazoo_publish("pushes", $avp(push_routing_key), $avp(push_payload));
+}
+
+route[PUSHER_ATTEMPT_REGISTRATION]
+{
+    if (!is_method("REGISTER")) {
+       return;
+    }
+
+    if($hdr(X-Token-Reg) != $null) {
+        if($sht(push_cache=>$hdr(X-Token-Reg)) != $null) {
+           $var(password) = $null;
+           $sht(push_cache=>$hdr(X-Token-Reg)) = $null;
+           xlog("L_INFO", "$ci|pusher|registration with x-token-reg\n");
+           route(SAVE_LOCATION);
+        } else {
+           xlog("L_INFO", "$ci|pusher|registration x-token-reg '$hdr(X-Token-Reg)' from header was not found\n");
+        }
+    }
+
+    if($(sel(contact.uri){uri.param,x-token-reg}) != "") {
+        if($sht(push_cache=>$(sel(contact.uri){uri.param,x-token-reg})) != $null) {
+           $var(password) = $null;
+           $sht(push_cache=>$(sel(contact.uri){uri.param,x-token-reg})) = $null;
+           xlog("L_INFO", "$ci|pusher|registration with x-token-reg\n");
+           route(SAVE_LOCATION);
+        } else {
+           xlog("L_INFO", "$ci|pusher|registration x-token-reg from contact '$(sel(contact.uri){uri.param,x-token-reg})' was not found\n");
+        }
+    }
+
 }
 
 route[PUSHER_ON_REGISTRATION]
 {
     if($(xavp(ulattrs=>custom_channel_vars){kz.json,Pusher-Application}{s.len}) > 0 &&  $var(Status) == "Registered") {
-       xlog("L_INFO", "$ci| pusher ON_REGISTRATION - $(xavp(ulattrs=>custom_channel_vars){kz.json,Pusher-Application})\n");
-       ts_append("location", "$tu");
+       if($sht(push_cache=>$(tu{s.tolower})) != $null) {
+           xlog("L_INFO", "$ci|pusher|device registered, delivering the call\n");
+           $var(ref) = $sht(push_cache=>$(tu{s.tolower}));
+           $sht(push_cache=>$(tu{s.tolower})) = $null;
+           $var(t_index) = $(var(ref){param.value,index}{s.int});
+           $var(t_label) = $(var(ref){param.value,label}{s.int});
+           t_continue("$var(t_index)", "$var(t_label)", "PUSHER_DELIVER_CALL");
+       }
+    }
+}
+
+route[PUSHER_DELIVER_CALL]
+{
+    if(lookup("location", "$hdr(X-KAZOO-AOR)") > 0) {
+       route(PUSHER_TO_EXTERNAL_RELAY);
+    } else {
+       t_reply(486, "Failed to lookup after resume");
     }
 }