######## Generic Hash Table container in shared memory ########
modparam("htable", "htable", "failover=>size=16;autoexpire=120")

#!trydef KZ_DISPATCHER_PROBE_MODE 1
#!trydef DISPATCHER_ADD_SERVERS 1
#!trydef DISPATCHER_ALG 0
#!trydef KZ_DISPATCHER_ADD_FLAGS 10

kazoo.dispatcher_auto_add = DISPATCHER_ADD_SERVERS descr "adds media servers reported by ecallmgr"
kazoo.dispatcher_algorithm = DISPATCHER_ALG descr "dispatcher algorithm to use"

####### Dispatcher module  ########
loadmodule "dispatcher.so"
modparam("dispatcher", "db_url", "KAZOO_DB_URL")
modparam("dispatcher", "flags", 2)
modparam("dispatcher", "use_default", 0)
modparam("dispatcher", "force_dst", 1)
modparam("dispatcher", "dst_avp", "$avp(ds_dst)")
modparam("dispatcher", "attrs_avp", "$avp(ds_attrs)")
modparam("dispatcher", "grp_avp", "$avp(ds_grp)")
modparam("dispatcher", "cnt_avp", "$avp(ds_cnt)")
modparam("dispatcher", "hash_pvar", "$avp(ds_grp)")
modparam("dispatcher", "setid_pvname", "$var(setid)")
modparam("dispatcher", "ds_ping_method", "OPTIONS")
modparam("dispatcher", "ds_ping_interval", 10)
modparam("dispatcher", "ds_probing_threshold", 3)
modparam("dispatcher", "ds_probing_mode", KZ_DISPATCHER_PROBE_MODE)
modparam("dispatcher", "ds_ping_reply_codes", "501,403,404,400,200")
modparam("dispatcher", "ds_ping_from", "sip:sipcheck@MY_HOSTNAME")

#!import_file "dispatcher-network-params.cfg"

## Dispatcher Groups:
## 1 - Primary media servers
## 2 - Backup media servers
## 3 - Alternate media server IPs (used only for classification)
## 10 - Presence servers (if not locally handled)
## 20 - Registrar servers (if not locally handled)


####### RTIMER module for dispatcher reload ##########
#!ifndef RTIMER_LOADED
loadmodule "rtimer.so"
#!trydef RTIMER_LOADED
#!endif
modparam("rtimer", "timer", "name=dispatcher_reload;interval=20;mode=1;")
modparam("rtimer", "exec", "timer=dispatcher_reload;route=DISPATCHER_RELOAD")

####### Dispatcher Logic ########
route[DISPATCHER_CLASSIFY_SOURCE]
{
    #!import_file "dispatcher-network-classify.cfg"
    
    if (is_myself("$ou")) {
        xlog("L_INFO", "$ci|log|original R-URI ($ou) is this proxy, treating as external sources\n");
    } else if (
      ds_is_from_list(1, 3) ||
      ds_is_from_list(2, 3) ||
      ds_is_from_list(3, 3) ||
      ds_is_from_list(10, 3) ||
      ds_is_from_list(20, 3)
    ) {
        xlog("L_INFO", "$ci|log|originated from internal sources\n");

        setflag(FLAG_INTERNALLY_SOURCED);
    } else {
        xlog("L_INFO", "$ci|log|originated from external sources\n");
    }
}

# Take the routes from dispatcher - hash over callid
# If prefered route defined, reorder the destionations
route[DISPATCHER_FIND_ROUTES]
{
    if ($sht(failover=>$ci::current) != $null) {
        $du = $sht(failover=>$ci::current);
        return;
    }

    $var(ds_primary_group) = 1;
    $var(ds_backup_group) = 2;
    
    #!ifndef PRESENCE_ROLE
    if (is_method("SUBSCRIBE")) {
        $var(ds_primary_group) = 10;
        $var(ds_backup_group) = 11;
    }
    #!endif

    #!ifndef REGISTRAR_ROLE
    if (is_method("REGISTER")) {
        $var(ds_primary_group) = 20;
        $var(ds_backup_group) = 21;
    }
    #!endif

    #!ifdef FAST_PICKUP_ROLE
    route(FAST_PICKUP_ATTEMPT);
    #!endif

    #!import_file "dispatcher-network-find.cfg"

    $var(ds_group) = $var(ds_primary_group);

    if (!ds_select_dst("$var(ds_primary_group)", "$sel(cfg_get.kazoo.dispatcher_algorithm)") || $(avp(ds_dst)[0]) == $null) {
        # we selected from primary group, try again in backup group
        if (!ds_select_dst("$var(ds_backup_group)", "$sel(cfg_get.kazoo.dispatcher_algorithm)") || $(avp(ds_dst)[0]) == $null) {
             xlog("L_WARN", "$ci|end|no servers available in primary or backup group\n");
             sl_send_reply("480", "All servers busy");
             exit;
        } else {
            $var(ds_group) = $var(ds_backup_group);
        }
    }

    $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) {
        $var(prefered_route) = $sht(redirects=>$var(redirect));
        xlog("L_INFO", "$ci|log|found redirect for $var(redirect)\n");
        if (route(DISPATCHER_REORDER_ROUTES)) {
            $avp(AVP_REDIRECT_KEY) = $var(redirect);
        }
    } 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;
        }
    }
}

route[DISPATCHER_REORDER_ROUTES]
{
    $var(i) = 0;
    $var(found) = 0;
    while($(avp(ds_dst)[$var(i)]) != $null) {
        if($(avp(ds_dst)[$var(i)]) != $var(prefered_route)) {
            $avp(tmp_ds_dst) = $(avp(ds_dst)[$var(i)]);
        } else {
            $var(found) = 1;
        }

        $var(i) = $var(i) + 1;
    }

    if (!$var(found) && $var(ds_group) == $var(ds_primary_group) && ds_select_dst("$var(ds_backup_group)", "0")) {
        $var(i) = 0;
        while($(avp(ds_dst)[$var(i)]) != $null) {
            if($(avp(ds_dst)[$var(i)]) == $var(prefered_route)) {
                xlog("L_INFO", "$ci|log|found associated media server in backup list\n");
                $var(found) = 1;
                break;
            }
            $var(i) = $var(i) + 1;
        }
    }

    if ($var(found)) {
        xlog("L_INFO", "$ci|log|re-ordering the dispatcher list to maintain association with $var(prefered_route)\n");

        $(avp(ds_dst)[*]) = $null;

        $var(i) = 0;
        while($(avp(tmp_ds_dst)[$var(i)]) != $null) {
            $avp(ds_dst) = $(avp(tmp_ds_dst)[$var(i)]);

            $var(i) = $var(i) + 1;
        }
        $avp(ds_dst) = $var(prefered_route);
        $du = $var(prefered_route);

        $(avp(tmp_ds_dst)[*]) = $null;
    } else {
        xlog("L_INFO", "$ci|log|associated media server $var(prefered_route) is inactive, moving to $rd\n");
        return -1;
    }

    return 1;
}

route[DISPATCHER_NEXT_ROUTE]
{
    $var(failover_count) = $sht(failover=>$ci::counts);

    if ($sht(failover=>$ci::counts) == $null) {
        $var(i) = 0;
        while($(avp(ds_dst)[$var(i)]) != $null) {
            $sht(failover=>$ci[$var(i)]) = $(avp(ds_dst)[$var(i)]);
            $var(i) = $var(i) + 1;
            if ($var(i) >= 3) {
                break;
            }
        }
        $sht(failover=>$ci::counts) = $var(i);
        $var(failover_count) = $var(i);
    }

    # try to find a new media server to send the call to
    if ($var(failover_count) > 1) {
        $var(failover_count) = $var(failover_count) - 1;

        $du = $sht(failover=>$ci[$var(failover_count)]);
        $sht(failover=>$ci::counts) = $var(failover_count);
        $sht(failover=>$ci::current) = $du;

        xlog("L_INFO", "$ci|log|remaining failed retry attempts: $var(failover_count)\n");
        xlog("L_INFO", "$ci|log|routing call to next media server $du\n");

        setflag(FLAG_SKIP_NAT_CORRECTION);

        # reset the final reply timer
        $avp(final_reply_timer) = 3;

        t_on_reply("INTERNAL_REPLY");

        t_on_failure("INTERNAL_FAULT");

        # relay the request to the new media server
        route(EXTERNAL_TO_INTERNAL_RELAY);

        exit();
    }
}

event_route[dispatcher:dst-down]
{
    xlog("L_ERR", "Destination down: $ru\n");
}

event_route[dispatcher:dst-up]
{
    xlog("L_WARNING", "Destination up: $ru\n");
}


route[DISPATCHER_CHECK_MEDIA_SERVER]
{
   if(@cfg_get.kazoo.dispatcher_auto_add == 1) {
       $var(SetId) = 1;
       if($var(Zone) != "MY_AMQP_ZONE") {
          $var(SetId) = 2;
       }
       $var(flags) = KZ_DISPATCHER_ADD_FLAGS;
       $var(attrs) = $_s(zone=$var(Zone);profile=$var(MediaProfile);idx=$(var(MediaUrl){s.corehash, MEDIA_SERVERS_HASH_SIZE});node=$var(MediaName));
       #!import_file "dispatcher-custom-media-check.cfg"
       sql_query("exec", "KZQ_CHECK_MEDIA_SERVER_INSERT");
       if($sqlrows(exec) > 0) {
          $shv(dispatcher_reload) = 1;
          return 1;
       }
   }
   return 0;
}

route[DISPATCHER_RELOAD]
{
   if($shv(dispatcher_reload) == 1) {
      xlog("L_WARNING", "reloading dispatcher table\n");
      ds_reload();
   };
   $shv(dispatcher_reload) = 0;  
}

route[DISPATCHER_STATUS]
{
    jsonrpc_exec('{"jsonrpc": "2.0", "method": "dispatcher.list", "id": 1}');
    $var(Sets) = $(jsonrpl(body){kz.json, result.NRSETS});
    $var(i) = 0;
    $var(ds_groups_json)="";
    $var(Sep1) = "";
    while($var(i) < $var(Sets)) {
       $var(Set) = $(jsonrpl(body){kz.json, result.RECORDS[$var(i)].SET});
       $var(SetCount) = $(var(Set){kz.json.count,TARGETS});
       $var(Sep2)="";
       $var(ds_group_json)="";
       $var(c) = 0;
       while($var(c) < $var(SetCount)) {
           $var(Dest) = $(var(Set){kz.json,TARGETS[$var(c)].DEST});
           $var(record) = $_s("$(var(Dest){kz.json,URI})" : {"destination" : "$(var(Dest){kz.json,URI})", "flags" : "$(var(Dest){kz.json,FLAGS})", "priority" : $(var(Dest){kz.json,PRIORITY}), "attrs" : "$(var(Dest){kz.json,ATTRS.BODY})"});
           $var(ds_group_json) = $var(ds_group_json) + $var(Sep2) + $var(record);
           $var(Sep2) = ",";
           $var(c) = $var(c) + 1;
       }
        $var(ds_groups_json) = $var(ds_groups_json) + $var(Sep1) + $_s("$(var(Set){kz.json,ID})" : { $var(ds_group_json) });
        $var(Sep1)=", ";
        $var(i) = $var(i) + 1;
    }
}

# vim: tabstop=4 softtabstop=4 shiftwidth=4 expandtab