bb_boxc.c

The Kannel Open Source WAP and SMS gateway works as both an SMS gateway, for implementing keyword b

         
        if ((boxc_id = cfg_get(grp, octstr_imm("smsbox-id"))) == NULL) { 
            grp_dump(grp); 
            panic(0,"'smsbox-route' group without valid 'smsbox-id' directive!"); 
        }

        /*
         * If smsc-ids are given, then any message comming from the specified
         * smsc-id will be routed to this smsbox instance.
         * If shortcuts are given, then any message with receiver number 
         * matching those will be routed to this smsbox instance.
         */
        smsc_ids = cfg_get(grp, octstr_imm("smsc-ids"));
        shortcuts = cfg_get(grp, octstr_imm("shortcuts"));

        /* now parse the smsc-ids and shortcuts semicolon separated list */
        if (smsc_ids) {
            items = octstr_split(smsc_ids, octstr_imm(";"));
            for (i = 0; i < list_len(items); i++) {
                Octstr *item = list_get(items, i);
                octstr_strip_blanks(item);

                debug("bb.boxc",0,"Adding smsbox routing to id <%s> for smsc id <%s>",
                      octstr_get_cstr(boxc_id), octstr_get_cstr(item));

                if (!dict_put_once(smsbox_by_smsc, item, octstr_duplicate(boxc_id)))
                    panic(0, "Routing for smsc-id <%s> already exists!",
                          octstr_get_cstr(item));
            }
            list_destroy(items, octstr_destroy_item);
            octstr_destroy(smsc_ids);
        }

        if (shortcuts) {
            items = octstr_split(shortcuts, octstr_imm(";"));
            for (i = 0; i < list_len(items); i++) {
                Octstr *item = list_get(items, i);
                octstr_strip_blanks(item);

                debug("bb.boxc",0,"Adding smsbox routing to id <%s> for receiver no <%s>",
                      octstr_get_cstr(boxc_id), octstr_get_cstr(item));
            
                if (!dict_put_once(smsbox_by_receiver, item, octstr_duplicate(boxc_id)))
                    panic(0, "Routing for receiver no <%s> already exists!",
                          octstr_get_cstr(item));
            }
            list_destroy(items, octstr_destroy_item);
            octstr_destroy(shortcuts);
        }
        octstr_destroy(boxc_id);
    }

    list_destroy(list, NULL);
}


/*-------------------------------------------------------------
 * public functions
 *
 * SMSBOX
 */

int smsbox_start(Cfg *cfg)
{
    CfgGroup *grp;
    
    if (smsbox_running) return -1;

    debug("bb", 0, "starting smsbox connection module");

    grp = cfg_get_single_group(cfg, octstr_imm("core"));
    if (cfg_get_integer(&smsbox_port, grp, octstr_imm("smsbox-port")) == -1) {
	    error(0, "Missing smsbox-port variable, cannot start smsboxes");
	    return -1;
    }
#ifdef HAVE_LIBSSL
    cfg_get_bool(&smsbox_port_ssl, grp, octstr_imm("smsbox-port-ssl"));
#endif /* HAVE_LIBSSL */

    if (smsbox_port_ssl)
        debug("bb", 0, "smsbox connection module is SSL-enabled");
        
    if (cfg_get_integer(&smsbox_max_pending, grp, octstr_imm("smsbox-max-pending")) == -1) {
        smsbox_max_pending = SMSBOX_MAX_PENDING;
        info(0, "BOXC: 'smsbox-max-pending' not set, using default (%ld).", smsbox_max_pending);
    }
    
    smsbox_list = list_create();	/* have a list of connections */
    smsbox_list_rwlock = gw_rwlock_create();
    if (!boxid)
        boxid = counter_create();

    /* the smsbox routing specific inits */
    smsbox_by_id = dict_create(10, NULL);  /* and a hash directory of identified */
    smsbox_by_smsc = dict_create(30, (void(*)(void *)) octstr_destroy);
    smsbox_by_receiver = dict_create(50, (void(*)(void *)) octstr_destroy);

    /* load the defined smsbox routing rules */
    init_smsbox_routes(cfg);

    list_add_producer(outgoing_sms);
    list_add_producer(smsbox_list);

    smsbox_running = 1;
    
    if ((sms_dequeue_thread = gwthread_create(sms_to_smsboxes, NULL)) == -1)
 	    panic(0, "Failed to start a new thread for smsbox routing");

    if (gwthread_create(smsboxc_run, (void *)smsbox_port) == -1)
	    panic(0, "Failed to start a new thread for smsbox connections");

    return 0;
}


int smsbox_restart(Cfg *cfg)
{
    if (!smsbox_running) return -1;
    
    /* send new config to clients */

    return 0;
}



/* WAPBOX */

int wapbox_start(Cfg *cfg)
{
    CfgGroup *grp;

    if (wapbox_running) return -1;

    debug("bb", 0, "starting wapbox connection module");
    
    grp = cfg_get_single_group(cfg, octstr_imm("core"));
    
    if (cfg_get_integer(&wapbox_port, grp, octstr_imm("wapbox-port")) == -1) {
	    error(0, "Missing wapbox-port variable, cannot start WAP");
	    return -1;
    }
#ifdef HAVE_LIBSSL
    cfg_get_bool(&wapbox_port_ssl, grp, octstr_imm("wapbox-port-ssl"));
#endif /* HAVE_LIBSSL */
  
    box_allow_ip = cfg_get(grp, octstr_imm("box-allow-ip"));
    if (box_allow_ip == NULL)
    	box_allow_ip = octstr_create("");
    box_deny_ip = cfg_get(grp, octstr_imm("box-deny-ip"));
    if (box_deny_ip == NULL)
    	box_deny_ip = octstr_create("");
    if (box_allow_ip != NULL && box_deny_ip == NULL)
	    info(0, "Box connection allowed IPs defined without any denied...");
    
    wapbox_list = list_create();	/* have a list of connections */
    list_add_producer(outgoing_wdp);
    if (!boxid)
        boxid = counter_create();

    if (gwthread_create(wdp_to_wapboxes, NULL) == -1)
 	    panic(0, "Failed to start a new thread for wapbox routing");
 
    if (gwthread_create(wapboxc_run, (void *)wapbox_port) == -1)
	    panic(0, "Failed to start a new thread for wapbox connections");

    wapbox_running = 1;
    return 0;
}


Octstr *boxc_status(int status_type)
{
    Octstr *tmp;
    char *lb, *ws;
    int i, boxes, para = 0;
    time_t orig, t;
    Boxc *bi;

    orig = time(NULL);

    /*
     * XXX: this will cause segmentation fault if this is called
     *    between 'destroy_list and setting list to NULL calls.
     *    Ok, this has to be fixed, but now I am too tired.
     */
    
    if ((lb = bb_status_linebreak(status_type))==NULL)
	    return octstr_create("Un-supported format");

    if (status_type == BBSTATUS_HTML)
	    ws = "&nbsp;&nbsp;&nbsp;&nbsp;";
    else if (status_type == BBSTATUS_TEXT)
	    ws = "    ";
    else
	    ws = "";

    if (status_type == BBSTATUS_HTML || status_type == BBSTATUS_WML)
	    para = 1;
    
    if (status_type == BBSTATUS_XML) {
        tmp = octstr_create ("");
        octstr_append_cstr(tmp, "<boxes>\n\t");
    }
    else
        tmp = octstr_format("%sBox connections:%s", para ? "<p>" : "", lb);
    boxes = 0;
    
    if (wapbox_list) {
	    list_lock(wapbox_list);
	    for(i=0; i < list_len(wapbox_list); i++) {
	        bi = list_get(wapbox_list, i);
	        if (bi->alive == 0)
		        continue;
	        t = orig - bi->connect_time;
            if (status_type == BBSTATUS_XML)
	            octstr_format_append(tmp,
		        "<box>\n\t\t<type>wapbox</type>\n\t\t<IP>%s</IP>\n"
                "\t\t<status>on-line %ldd %ldh %ldm %lds</status>\n"
                "\t\t<ssl>%s</ssl>\n\t</box>\n",
				octstr_get_cstr(bi->client_ip),
				t/3600/24, t/3600%24, t/60%60, t%60,
#ifdef HAVE_LIBSSL
                conn_get_ssl(bi->conn) != NULL ? "yes" : "no"
#else 
                "not installed"
#endif
                );
            else
	            octstr_format_append(tmp,
		        "%swapbox, IP %s (on-line %ldd %ldh %ldm %lds) %s %s",
				ws, octstr_get_cstr(bi->client_ip),
				t/3600/24, t/3600%24, t/60%60, t%60, 
#ifdef HAVE_LIBSSL
                conn_get_ssl(bi->conn) != NULL ? "using SSL" : "",
#else
                "",
#endif 
                lb);
	            boxes++;
	       }
	       list_unlock(wapbox_list);
        }
        if (smsbox_list) {
            gw_rwlock_rdlock(smsbox_list_rwlock);
	    for(i=0; i < list_len(smsbox_list); i++) {
	        bi = list_get(smsbox_list, i);
	        if (bi->alive == 0)
		        continue;
	        t = orig - bi->connect_time;
            if (status_type == BBSTATUS_XML)
	            octstr_format_append(tmp, "<box>\n\t\t<type>smsbox</type>\n"
                    "\t\t<id>%s</id>\n\t\t<IP>%s</IP>\n"
                    "\t\t<queue>%ld</queue>\n"
                    "\t\t<status>on-line %ldd %ldh %ldm %lds</status>\n"
                    "\t\t<ssl>%s</ssl>\n\t</box>",
                    (bi->boxc_id ? octstr_get_cstr(bi->boxc_id) : ""),
		            octstr_get_cstr(bi->client_ip),
		            list_len(bi->incoming) + dict_key_count(bi->sent),
		            t/3600/24, t/3600%24, t/60%60, t%60,
#ifdef HAVE_LIBSSL
                    conn_get_ssl(bi->conn) != NULL ? "yes" : "no"
#else 
                    "not installed"
#endif
                    );
            else
                octstr_format_append(tmp, "%ssmsbox:%s, IP %s (%ld queued), (on-line %ldd %ldh %ldm %lds) %s %s",
                    ws, (bi->boxc_id ? octstr_get_cstr(bi->boxc_id) : "(none)"), 
                    octstr_get_cstr(bi->client_ip), list_len(bi->incoming) + dict_key_count(bi->sent),
		            t/3600/24, t/3600%24, t/60%60, t%60, 
#ifdef HAVE_LIBSSL
                    conn_get_ssl(bi->conn) != NULL ? "using SSL" : "",
#else
                    "",
#endif 
                    lb);
	       boxes++;
	    }
	    gw_rwlock_unlock(smsbox_list_rwlock);
    }
    if (boxes == 0 && status_type != BBSTATUS_XML) {
	    octstr_destroy(tmp);
	    tmp = octstr_format("%sNo boxes connected", para ? "<p>" : "");
    }
    if (para)
	    octstr_append_cstr(tmp, "</p>");
    if (status_type == BBSTATUS_XML)
        octstr_append_cstr(tmp, "</boxes>\n");
    else
        octstr_append_cstr(tmp, "\n\n");
    return tmp;
}


int boxc_incoming_wdp_queue(void)
{
    int i, q = 0;
    Boxc *boxc;
    
    if (wapbox_list) {
	    list_lock(wapbox_list);
	    for(i=0; i < list_len(wapbox_list); i++) {
	        boxc = list_get(wapbox_list, i);
	        q += list_len(boxc->incoming);
	    }
	    list_unlock(wapbox_list);
    }
    return q;
}


void boxc_cleanup(void)
{
    octstr_destroy(box_allow_ip);
    octstr_destroy(box_deny_ip);
    box_allow_ip = NULL;
    box_deny_ip = NULL;
    counter_destroy(boxid);
    boxid = NULL;
}


/*
 * Route the incoming message to one of the following input queues:
 *   a specific smsbox conn
 *   a random smsbox conn if no shortcut routing and msg->sms.boxc_id match
 *
 * BEWARE: All logic inside here should be fast, hence speed processing
 * optimized, because every single MO message passes this function and we 
 * have to ensure that no unncessary overhead is done.
 */
int route_incoming_to_boxc(Msg *msg)
{
    Boxc *bc = NULL, *best = NULL;
    Octstr *s, *r;
    long len, b, i;
    int full_found = 0;

    s = r = NULL;
    gw_assert(msg_type(msg) == sms);

    /* msg_dump(msg, 0); */

    /* 
     * We have a specific route to pass this msg to smsbox-id 
     * Lookup the connection in the dictionary.
     */
    gw_rwlock_rdlock(smsbox_list_rwlock);
    if (list_len(smsbox_list) == 0) {
        gw_rwlock_unlock(smsbox_list_rwlock);
    	warning(0, "smsbox_list empty!");
        if (max_incoming_sms_qlength < 0 || max_incoming_sms_qlength > list_len(incoming_sms)) {
            list_produce(incoming_sms, msg);
	    return 0;
         }
         else
             return -1;
    }

    if (octstr_len(msg->sms.boxc_id) > 0) {
        
        bc = dict_get(smsbox_by_id, msg->sms.boxc_id);
        if (bc == NULL) {
            /*
             * something is wrong, this was the smsbox connection we used
             * for sending, so it seems this smsbox is gone
             */
            warning(0,"Could not route message to smsbox id <%s>, smsbox is gone!",
                  octstr_get_cstr(msg->sms.boxc_id));
        }
    }
    else {
        /*
         * Check if we have a "smsbox-route" for this msg.
         * Where the shortcut route has a higher priority then the smsc-id rule.
         */
        s = (msg->sms.smsc_id ? dict_get(smsbox_by_smsc, msg->sms.smsc_id) : NULL);
        r = (msg->sms.receiver ? dict_get(smsbox_by_receiver, msg->sms.receiver) : NULL);
        bc = r ? dict_get(smsbox_by_id, r) : (s ? dict_get(smsbox_by_id, s) : NULL);
    }

    /* check if we found our routing */
    if (bc != NULL) {
        if (max_incoming_sms_qlength < 0 || max_incoming_sms_qlength > list_len(bc->incoming)) {
            list_produce(bc->incoming, msg);
            gw_rwlock_unlock(smsbox_list_rwlock);
            return 1; /* we are done */
        }
        else {
            gw_rwlock_unlock(smsbox_list_rwlock);
            return -1;
        }
    }
    else if (s != NULL || r != NULL || octstr_len(msg->sms.boxc_id) > 0) {
        gw_rwlock_unlock(smsbox_list_rwlock);
        /*
         * we have routing defined, but no smsbox connected at the moment.
         * put msg into global incoming queue and wait until smsbox with
         * such boxc_id connected.
         */
        if (max_incoming_sms_qlength < 0 || max_incoming_sms_qlength > list_len(incoming_sms)) {
            list_produce(incoming_sms, msg);
            return 0;
        }
        else
            return -1;
    }

    /*
     * ok, none of the routing things applied previously, so route it to
     * a random smsbox.
     */

    /* take random smsbox from list, and then check all smsboxes
     * and select the one with lowest load level - if tied, the first
     * one
     */
    len = list_len(smsbox_list);
    b = gw_rand() % len;

    for(i = 0; i < list_len(smsbox_list); i++) {
	bc = list_get(smsbox_list, (i+b) % len);

        if (bc->boxc_id != NULL || bc->routable == 0)
            bc = NULL;

        if (bc != NULL && max_incoming_sms_qlength > 0 &&
            list_len(bc->incoming) > max_incoming_sms_qlength) {
            full_found = 1;
            bc = NULL;
        }

        if ((bc != NULL && best != NULL && bc->load < best->load) ||
             (bc != NULL && best == NULL)) {
	    best = bc;
        }
    }

    if (best != NULL) {
        best->load++;
        list_produce(best->incoming, msg);
    }

    gw_rwlock_unlock(smsbox_list_rwlock);

    if (best == NULL && full_found == 0) {
	warning(0, "smsbox_list empty!");
        if (max_incoming_sms_qlength < 0 || max_incoming_sms_qlength > list_len(incoming_sms)) {
            list_produce(incoming_sms, msg);
	    return 0;
         }
         else
             return -1;
    }
    else if (best == NULL && full_found == 1)
        return -1;

    return 1;
}

static void sms_to_smsboxes(void *arg)
{
    Msg *newmsg, *startmsg, *msg;
    long i, len;
    int ret = -1;
    Boxc *boxc;

    list_add_producer(flow_threads);

    newmsg = startmsg = msg = NULL;

    while(bb_status != BB_DEAD) {

        if (newmsg == startmsg) {
            /* check if we are in shutdown phase */
            if (list_producer_count(smsbox_list) == 0)
                break;

            if (ret == 0 || ret == -1) {
                /* debug("", 0, "time to sleep"); */
                gwthread_sleep(60.0);
                /* debug("", 0, "wake up list len %ld", list_len(incoming_sms)); */
                /* shutdown ? */
                if (list_producer_count(smsbox_list) == 0 && list_len(smsbox_list) == 0)
                    break;
            }
            startmsg = msg = list_consume(incoming_sms);
            /* debug("", 0, "list_consume done 1"); */
            newmsg = NULL;
        }
        else {
            newmsg = msg = list_consume(incoming_sms);
        }

        if (msg == NULL)
            break;

        gw_assert(msg_type(msg) == sms);

        /* debug("bb.sms", 0, "sms_boxc_router: handling message (%p vs %p)",
	          msg, startmsg); */

        ret = route_incoming_to_boxc(msg);
        if (ret == 1)
            startmsg = newmsg = NULL;
        else if (ret == -1) {
            list_produce(incoming_sms, msg);
        }
    }

    gw_rwlock_rdlock(smsbox_list_rwlock);
    len = list_len(smsbox_list);
    for (i=0; i < len; i++) {
        boxc = list_get(smsbox_list, i);
        list_remove_producer(boxc->incoming);
    }
    gw_rwlock_unlock(smsbox_list_rwlock);

    list_remove_producer(flow_threads);
}