#!substdef "!PRESENCE_PROCESS_SLEEP!100000!g"
modparam("mqueue", "mqueue", "name=presence")
modparam("rtimer", "timer", "name=presenceMaster;interval=PRESENCE_PROCESS_SLEEPu;mode=1;")
modparam("rtimer", "exec", "timer=presenceMaster;route=PRESENCE_MASTER_PROCESS")
modparam("mqueue", "mqueue", "name=presenceWorker0")
modparam("rtimer", "timer", "name=presenceWorker0;interval=1u;mode=1;")
modparam("rtimer", "exec", "timer=presenceWorker0;route=PRESENCE_WORKER_PROCESS")
modparam("mqueue", "mqueue", "name=presenceWorker1")
modparam("rtimer", "timer", "name=presenceWorker1;interval=1u;mode=1;")
modparam("rtimer", "exec", "timer=presenceWorker1;route=PRESENCE_WORKER_PROCESS")
...
modparam("mqueue", "mqueue", "name=presenceWorkern")
modparam("rtimer", "timer", "name=presenceWorkern;interval=1u;mode=1;")
modparam("rtimer", "exec", "timer=presenceWorkern;route=PRESENCE_WORKER_PROCESS")
...
route {
	...
	# Some logic to determine this is a presence request (within or without dialog)
	$var(queue) = "presence";
	route(PRESENCE_ENQUEUE);
	...
}
...
route[PRESENCE_ENQUEUE] {
        xlog("L_WARN", "$rm: route[PRESENCE_ENQUEUE]\n");

        if (!t_suspend()) {
                t_reply("500", "Server Internal Error");
                xlog("L_ERR", "Failed to suspend transaction for $rm\n");
                exit;
        }

        xlog("L_INFO", "Suspended transaction for $rm [$T(id_index):$T(id_label)]\n");

        if (!mq_add("$var(queue)", "$T(id_index)", "$T(id_label)")) {
                t_reply("500", "Server Internal Error");
                xlog("L_ERR", "Failed to queue transaction for $rm [$T(id_index):$T(id_label)] on $var(queue)\n");
                exit;
        }

        exit;
}

route[PRESENCE_MASTER_PROCESS] {
        xlog("L_INFO", "Running PRESENCE_MASTER_PROCESS\n");

        while (mq_fetch("presence")) {
                $var(id_index) = (int) $mqk(presence);
                $var(id_label) = (int) $mqv(presence);
                xlog("L_INFO", "presence: Found queued transaction [$var(id_index):$var(id_label)]\n");
                t_continue("$var(id_index)", "$var(id_label)", "PRESENCE_DISTRIBUTE");
        }
}

route[PRESENCE_DISTRIBUTE] {
        xlog("L_WARN", "$rm: route[PRESENCE_DISTRIBUTE]\n");

        # Some algorithm to distribute traffic across queues...
	# Perhaps on request type (so we have a NOTIFIER that
	# sends NOTIFY requests in order), perhaps on some form
	# of hash...  I am experimenting with this...
        $var(queue) = "presenceWorker" + $var(queue_number);
        xlog("L_INFO", "Adding to queue: $var(queue)\n");
        route(PRESENCE_ENQUEUE);
}

route[PRESENCE_WORKER_PROCESS] {
        lock("pres");
        $var(pres) = $shv(pres);
        $shv(pres) = $shv(pres) + 1;
        unlock("pres");

        $var(queue) = "presenceWorker" + $var(pres);
        xlog("L_WARN", "Starting process: $var(queue) (pid: $pp)\n");

        while (1) {
                while (mq_fetch($var(queue))) {
                        $var(id_index) = (int) $mqk($var(queue));
                        $var(id_label) = (int) $mqv($var(queue));
                        xlog("L_WARN", "$var(queue): found queued transaction [$var(id_index):$var(id_label)]\n");
                        t_continue("$var(id_index)", "$var(id_label)", "PRESENCE");
                }
                usleep(PRESENCE_PROCESS_SLEEP);
        }
}

route[PRESENCE] {
        xlog("L_WARN", "$rm: route[PRESENCE]: $var(queue)\n");

        if (is_method("NOTIFY")) {
                xlog("L_INFO", "Sending NOTIFY to RLS\n");
                rls_handle_notify();
        } else if (is_method("PUBLISH")) {
                xlog("L_INFO", "Sending PUBLISH to Presence\n");
                handle_publish();
        } else if (is_method("SUBSCRIBE")) {
                xlog("L_INFO", "Sending SUBSCRIBE to RLS\n");
                $var(ret_code) = rls_handle_subscribe();
                if ($var(ret_code) == 10) {
                        xlog("L_INFO", "  SUBSCRIBE not for RLS - sending to Presence\n");
                        handle_subscribe();
                }
        } else {
                xlog("L_ERR", "Received non-(NOTIFY|PUBLISH|SUBSCRIBE) request from presence queue\n");
                t_reply("500", "Server Internal Error");
        }
	exit;
}
...