Re: [PATCH] MO concat

=== CVS/Entries
==================================================================
--- CVS/Entries	(revision 288)
+++ CVS/Entries	(local)
@@ -30,4 +30,4 @@
 /configure.in/1.173/Wed Apr 11 17:06:10 2007//
 /Makefile.in/1.80/Sun Apr 22 11:22:43 2007//
 /aclocal.m4/1.10/Sun Apr 22 11:22:43 2007//
-/ChangeLog/1.2887/Wed May  9 09:11:23 2007//
+/ChangeLog/1.2888/Tue May 15 18:13:56 2007//
=== doc/userguide/userguide.xml
==================================================================
--- doc/userguide/userguide.xml	(revision 288)
+++ doc/userguide/userguide.xml	(local)
@@ -1534,6 +1534,22 @@
         </entry>   
      </row>
 
+     <row><entry><literal>sms-combine-concatenated-mo</literal></entry>
+        <entry>boolean</entry>
+        <entry valign="bottom">
+        Whether Kannel should attempt to combine concatenated MO SMS
+        prior to passing them over to smsbox. Default is true
+        </entry>   
+     </row>
+
+     <row><entry><literal>sms-combine-concatenated-mo-timeout</literal></entry>
+        <entry>seconds</entry>
+        <entry valign="bottom">
+        How long to wait for all concatenated message parts to arrive before timeouting.
+        Default 1800 seconds.
+        </entry>   
+     </row>
+
   </tbody>
   </tgroup>
  </table>
=== gw/bb_smscconn.c
==================================================================
--- gw/bb_smscconn.c	(revision 288)
+++ gw/bb_smscconn.c	(local)
@@ -106,6 +106,8 @@
 
 /* outgoing sms queue control */
 extern long max_outgoing_sms_qlength;
+/* incoming sms queue control */
+extern long max_incoming_sms_qlength;
 
 /* our own thingies */
 
@@ -133,11 +135,22 @@
  */
 Counter *split_msg_counter;
 
+/* Flag for handling concatenated incoming messages. */
+static int handle_concatenated_mo;
+/* How long to wait for message parts */
+static long concatenated_mo_timeout;
+/* Flag for return value of check_concat */
+enum {concat_error = -1, concat_complete = 0, concat_pending = 1, concat_none};
+
 /*
  * forward declaration
  */
 static long route_incoming_to_smsc(SMSCConn *conn, Msg *msg);
 
+static void init_concat_handler(void);
+static void shutdown_concat_handler(void);
+static int check_concatenation(Msg **msg, Octstr *smscid);
+static void clear_old_concat_parts(void);
 
 /*---------------------------------------------------------------------------
  * CALLBACK FUNCTIONS
@@ -386,11 +399,13 @@
 
     /* fix sms type if not set already */
     if (sms->sms.sms_type != report_mo)
-	sms->sms.sms_type = mo;
+        sms->sms.sms_type = mo;
 
     /* write to store (if enabled) */
-    if (store_save(sms) == -1)
-	return SMSCCONN_FAILED_TEMPORARILY;
+    if (store_save(sms) == -1) {
+        msg_destroy(sms);
+        return SMSCCONN_FAILED_TEMPORARILY;
+    }
 
     copy = msg_duplicate(sms);
 
@@ -400,6 +415,36 @@
      * Scope: internal routing (to smsc-ids)
      */
     if ((rc = route_incoming_to_smsc(conn, copy)) == -1) {
+        int ret;
+        /* Before routing to some box, do concat handling
+         * and replace copy as such.
+         */
+        if (handle_concatenated_mo && copy->sms.sms_type == mo) {
+            ret = check_concatenation(&copy, conn->id);
+            switch(ret) {
+            case concat_pending:
+                counter_increase(incoming_sms_counter); /* ?? */
+                if (conn != NULL)
+                    counter_increase(conn->received);
+                msg_destroy(sms);
+                return SMSCCONN_SUCCESS;
+            case concat_complete:
+                /* Combined sms received! save new one since it is now combined. */ 
+                msg_destroy(sms);
+                /* Change the sms. */
+                sms = msg_duplicate(copy);
+                break;
+            case concat_error:
+                /* failed to save, go away. */
+                msg_destroy(sms);
+                return SMSCCONN_FAILED_TEMPORARILY;
+            case concat_none:
+                break;
+            default:
+                panic(0, "Internal error: Unhandled concat result.");
+                break;
+            }
+        }
         /*
          * Now try to route the message to a specific smsbox
          * connection based on the existing msg->sms.boxc_id or
@@ -452,14 +497,16 @@
 {
     Msg *msg, *startmsg, *newmsg;
     long ret;
+    time_t concat_mo_check;
 
     gwlist_add_producer(flow_threads);
     gwthread_wakeup(MAIN_THREAD_ID);
 
     startmsg = newmsg = NULL;
     ret = SMSCCONN_SUCCESS;
+    concat_mo_check = time(NULL);
 
-    while(bb_status != BB_DEAD) {
+    while(bb_status != BB_SHUTDOWN && bb_status != BB_DEAD) {
 
 	if (newmsg == startmsg) {
             if (ret == SMSCCONN_QUEUED || ret == SMSCCONN_FAILED_QFULL) {
@@ -469,16 +516,22 @@
                 gwthread_sleep(sleep_time);
                 debug("bb.sms", 0, "sms_router: gwlist_len = %ld", gwlist_len(outgoing_sms));
             }
-            startmsg = msg = gwlist_consume(outgoing_sms);
+            startmsg = msg = gwlist_timed_consume(outgoing_sms, concatenated_mo_timeout);
             newmsg = NULL;
+        } else {
+            newmsg = msg = gwlist_timed_consume(outgoing_sms, concatenated_mo_timeout);
         }
-        else {
-            newmsg = msg = gwlist_consume(outgoing_sms);
+
+        if (difftime(time(NULL), concat_mo_check) > concatenated_mo_timeout) {
+            concat_mo_check = time(NULL);
+            clear_old_concat_parts();
         }
 
-        /* shutdown ? */
-        if (msg == NULL)
-            break;
+        /* shutdown or timeout */
+        if (msg == NULL) {
+            newmsg = startmsg = NULL;
+            continue;
+        }
 
         debug("bb.sms", 0, "sms_router: handling message (%p vs %p)",
                   msg, startmsg);
@@ -577,16 +630,16 @@
     else
         info(0, "SMS resend retry set to %ld.", sms_resend_retry);
 
+    if (cfg_get_bool(&handle_concatenated_mo, grp, octstr_imm("sms-combine-concatenated-mo")) == -1)
+        handle_concatenated_mo = 1; /* default is TRUE. */
+
+    if (cfg_get_integer(&concatenated_mo_timeout, grp, octstr_imm("sms-combine-concatenated-mo-timeout")) == -1)
+        concatenated_mo_timeout = 1800;
+
+    if (handle_concatenated_mo)
+        init_concat_handler();
+
     smsc_groups = cfg_get_multi_group(cfg, octstr_imm("smsc"));
-    /*
-    while(groups && (grp = gwlist_extract_first(groups)) != NULL) {
-        conn = smscconn_create(grp, 1); 
-        if (conn == NULL)
-            panic(0, "Cannot start with SMSC connection failing");
-        
-        gwlist_append(smsc_list, conn);
-    }
-    */
     gwlist_add_producer(smsc_list);
     for (i = 0; i < gwlist_len(smsc_groups) && 
         (grp = gwlist_get(smsc_groups, i)) != NULL; i++) {
@@ -820,6 +873,9 @@
     counter_destroy(split_msg_counter);
     gw_rwlock_destroy(&smsc_list_lock);
 
+    /* Stop concat handling */
+    shutdown_concat_handler();
+
     smsc_running = 0;
 }
 
@@ -1134,3 +1190,297 @@
     return -1; 
 }
 
+
+/*--------------------------------
+ * incoming concatenated messages handling
+ */
+
+typedef struct ConcatMsg {
+    int refnum;
+    int total_parts;
+    int num_parts;
+    time_t trecv;
+    Octstr *key; /* in dict. */
+    int ack;     /* set to the type of ack to send when deleting. */
+    /* array of parts */
+    Msg **parts;
+} ConcatMsg;
+
+static Dict *incoming_concat_msgs;
+static Mutex *concat_lock;
+static Msg *timeouted_message;
+
+static void destroy_concatMsg(void *x)
+{
+    int i;
+    ConcatMsg *msg = x;
+
+    gw_assert(msg);
+    for (i = 0; i < msg->total_parts; i++) {
+        if (msg->parts[i]) {
+            store_save_ack(msg->parts[i], msg->ack);
+            msg_destroy(msg->parts[i]);
+        }
+    }
+    gw_free(msg->parts);
+    octstr_destroy(msg->key);
+    gw_free(msg);
+}
+
+static void init_concat_handler(void)
+{
+    if (incoming_concat_msgs != NULL) /* already initialised? */
+        return;
+    incoming_concat_msgs = dict_create(max_incoming_sms_qlength > 0 ? max_incoming_sms_qlength : 1024, 
+                                       destroy_concatMsg);
+    concat_lock = mutex_create();
+    debug("bb.sms",0,"smsbox MO concatenated message handling enabled");
+}
+
+static void shutdown_concat_handler(void)
+{
+    if (incoming_concat_msgs == NULL)
+        return;
+    dict_destroy(incoming_concat_msgs);
+    mutex_destroy(concat_lock);
+
+    incoming_concat_msgs = NULL;
+    concat_lock = NULL;
+    debug("bb.sms",0,"smsbox MO concatenated message handling cleaned up");
+}
+
+static void clear_old_concat_parts(void)
+{
+    List *keys;
+    Octstr *key;
+
+    /* not initialised, go away */
+    if (incoming_concat_msgs != NULL)
+        return;
+
+    debug("bb.sms.splits", 0, "clear_old_concat_parts called");
+
+    /* Remove any pending messages that are too old. */
+    keys = dict_keys(incoming_concat_msgs);
+    while((key = gwlist_extract_first(keys)) != NULL) {
+        ConcatMsg *x;
+        Msg *msg;
+        int i, destroy = 1;
+
+        mutex_lock(concat_lock);
+        x = dict_get(incoming_concat_msgs, key);
+        octstr_destroy(key);
+        if (x == NULL || difftime(time(NULL), x->trecv) < concatenated_mo_timeout) {
+            mutex_unlock(concat_lock);
+            continue;
+        }
+        dict_remove(incoming_concat_msgs, x->key);
+        mutex_unlock(concat_lock);
+        warning(0, "Time-out waiting for concatenated message '%s'. Send message parts as is.",
+                octstr_get_cstr(x->key));
+        for (i = 0; i < x->total_parts && destroy == 1; i++) {
+            if (x->parts[i] == NULL)
+                continue;
+            msg = msg_duplicate(x->parts[i]);
+            mutex_lock(concat_lock);
+            timeouted_message = msg;
+            mutex_unlock(concat_lock);
+            store_save_ack(x->parts[i], ack_success);
+            switch(bb_smscconn_receive(NULL, msg)) {
+            case SMSCCONN_FAILED_REJECTED:
+            case SMSCCONN_SUCCESS:
+                msg_destroy(x->parts[i]);
+                x->parts[i] = NULL;
+                x->num_parts--;
+                break;
+            case SMSCCONN_FAILED_TEMPORARILY:
+            case SMSCCONN_FAILED_QFULL:
+            default:
+                /* oops put it back into dict and retry on next run */
+                store_save(x->parts[i]);
+                destroy = 0;
+                break;
+            }
+            mutex_lock(concat_lock);
+            timeouted_message = NULL;
+            mutex_unlock(concat_lock);
+        }
+        if (destroy) {
+            destroy_concatMsg(x);
+        } else {
+            ConcatMsg *x1;
+            mutex_lock(concat_lock);
+            x1 = dict_get(incoming_concat_msgs, x->key);
+            if (x1 != NULL) { /* oops we have new part */
+                int i;
+                if (x->total_parts != x1->total_parts) {
+                    /* broken handset, don't know what todo here??
+                     * for now just put old concatMsg into dict with
+                     * another key and it will be cleaned up on next run.
+                     */
+                    octstr_format_append(x->key, " %d", x->total_parts);
+                    dict_put(incoming_concat_msgs, x->key, x);
+                } else {
+                    for (i = 0; i < x->total_parts; i++) {
+                        if (x->parts[i] == NULL)
+                            continue;
+                        if (x1->parts[i] == NULL) {
+                            x1->parts[i] = x->parts[i];
+                            x->parts[i] = NULL;
+                        }
+                    }
+                    destroy_concatMsg(x);
+                }
+            } else {
+                dict_put(incoming_concat_msgs, x->key, x);
+            }
+            mutex_unlock(concat_lock);
+        }
+    }
+    gwlist_destroy(keys, octstr_destroy_item);
+}
+
+/* Checks if message is concatenated. Returns:
+ * - returns concat_complete if no concat parts, or message complete
+ * - returns concat_pending (and sets *pmsg to NULL) if parts pending
+ * - returns concat_error if store_save fails
+ */
+static int check_concatenation(Msg **pmsg, Octstr *smscid)
+{
+    Msg *msg = *pmsg;
+    int l, iel, refnum, pos, c, part, totalparts, i, sixteenbit;
+    Octstr *udh = msg->sms.udhdata, *key;
+    ConcatMsg *cmsg;
+    int ret = concat_complete;
+
+    /* ... module not initialised or there is no UDH. */
+    if (incoming_concat_msgs == NULL || (l = octstr_len(udh)) == 0)
+        return concat_none;
+
+    /* check if this is timeouted message */
+    mutex_lock(concat_lock);
+    if (timeouted_message == msg) {
+        mutex_unlock(concat_lock);
+        return concat_none;
+    }
+    mutex_unlock(concat_lock);
+
+    for (pos = 1, c = -1; pos < l - 1; pos += iel + 2) {
+        iel = octstr_get_char(udh, pos + 1);
+        if ((c = octstr_get_char(udh,pos)) == 0 || c == 8)
+            break;
+    }
+    if (pos >= l)  /* no concat UDH found. */
+        return concat_none;
+
+    /* c = 0 means 8 bit, c = 8 means 16 bit concat info */
+    sixteenbit = (c == 8);
+    refnum = (!sixteenbit) ? octstr_get_char(udh, pos + 2) :
+        (octstr_get_char(udh, pos + 2) << 8) | octstr_get_char(udh, pos + 3);
+    totalparts = octstr_get_char(udh, pos + 3 + sixteenbit);
+    part = octstr_get_char(udh, pos + 4 + sixteenbit);
+
+    if (part < 1 || part > totalparts) {
+        warning(0, "Invalid concatenation UDH [ref = %d] in message from %s!",
+                refnum, octstr_get_cstr(msg->sms.sender));
+        return concat_none;
+    }
+
+    debug("bb.sms.splits", 0, "Got part %d [ref %d, total parts %d] of message from %s. Dump follows:",
+          part, refnum,totalparts, octstr_get_cstr(msg->sms.sender));
+     
+    msg_dump(msg,0);
+     
+    key = octstr_format("%S %S %S %d", msg->sms.sender, msg->sms.receiver, smscid, refnum);
+    mutex_lock(concat_lock);
+    if ((cmsg = dict_get(incoming_concat_msgs, key)) == NULL) {
+        cmsg = gw_malloc(sizeof(*cmsg));
+        cmsg->refnum = refnum;
+        cmsg->total_parts = totalparts;
+        cmsg->num_parts = 0;
+        cmsg->key = octstr_duplicate(key);
+        cmsg->ack = ack_success;
+        cmsg->parts = gw_malloc(totalparts * sizeof(*cmsg->parts));
+        memset(cmsg->parts, 0, cmsg->total_parts * sizeof(*cmsg->parts)); /* clear it. */
+
+        dict_put(incoming_concat_msgs, key, cmsg);
+    }
+    octstr_destroy(key);
+
+    if (totalparts != cmsg->total_parts) {
+        /* totalparts in udh and cmsg not equal assume bad message */
+        error(0, "Totalparts in UDH doesn't match received before, "
+                "total parts <%d>:<%d> part %d, ref %d, from %s, to %s. Discarded!",
+                cmsg->total_parts, totalparts, part, refnum, octstr_get_cstr(msg->sms.sender), octstr_get_cstr(msg->sms.receiver));
+        mutex_unlock(concat_lock);
+        store_save_ack(msg, ack_success);
+        msg_destroy(msg);
+        *pmsg = msg = NULL;
+        return concat_error;
+    }
+
+    /* check if we have seen message part before... */
+    if (cmsg->parts[part - 1] != NULL) {	  
+        warning(0, "Duplicate message part %d, ref %d, from %s, to %s. Discarded!",
+                part, refnum, octstr_get_cstr(msg->sms.sender), octstr_get_cstr(msg->sms.receiver));
+        store_save_ack(msg, ack_success);
+        msg_destroy(msg); 
+        *pmsg = msg = NULL;
+    } else {
+        cmsg->parts[part -1] = msg;
+        cmsg->num_parts++;
+        /* always update receive time so we have it from last part and don't timeout */
+        cmsg->trecv = time(NULL);
+    }
+
+    if (cmsg->num_parts < cmsg->total_parts) {  /* wait for more parts. */
+        *pmsg = msg = NULL;
+        mutex_unlock(concat_lock);
+        return concat_pending;
+    }
+
+    /* we have all the parts: Put them together, mod UDH, return message. */
+    msg = msg_duplicate(cmsg->parts[0]);
+    uuid_generate(msg->sms.id); /* give it a new ID. */
+
+    debug("bb.sms.splits",0,"Received all concatenated message parts from %s, to %s, refnum %d",
+          octstr_get_cstr(msg->sms.sender), octstr_get_cstr(msg->sms.receiver), refnum);
+
+    for (i = 1; i < cmsg->total_parts; i++)
+        octstr_append(msg->sms.msgdata, cmsg->parts[i]->sms.msgdata);
+
+    /* Attempt to save the new one, if that fails, then reply with fail. */
+    if (store_save(msg) == -1) {	  
+        mutex_unlock(concat_lock);
+        msg_destroy(msg);
+        *pmsg = msg = NULL;
+        return concat_error;
+    } else 
+        *pmsg = msg; /* return the message part. */
+
+    /* Delete it from the queue and from the Dict. */
+    /* Note: dict_put with NULL value delete and destroy value */
+    dict_put(incoming_concat_msgs, cmsg->key, NULL);
+    mutex_unlock(concat_lock);
+
+    /* fix up UDH */
+    udh = msg->sms.udhdata;
+    l = octstr_len(udh);
+    for (pos = 1; pos < l - 1; pos += iel + 2) {
+        iel = octstr_get_char(udh, pos + 1);
+        if ((c = octstr_get_char(udh, pos)) == 0 || c == 8) {
+            octstr_delete(udh, pos, iel + 2);
+
+            if (octstr_len(udh) <= 1) /* no other UDH elements. */
+                octstr_delete(udh, 0, octstr_len(udh));
+            else
+                octstr_set_char(udh, 0, octstr_len(udh) - 1);
+            break;
+        }
+    }
+    debug("bb.sms.splits", 0, "Got full message [ref %d] of message from %s to %s. Dumping: ",
+          refnum, octstr_get_cstr(msg->sms.sender), octstr_get_cstr(msg->sms.receiver));
+    msg_dump(msg,0);
+
+    return ret;
+}
=== gwlib/CVS/Entries
==================================================================
--- gwlib/CVS/Entries	(revision 288)
+++ gwlib/CVS/Entries	(local)
@@ -45,8 +45,6 @@
 /http.c/1.248/Sun Jan 14 19:21:31 2007//
 /http.h/1.70/Sun Jan 14 19:21:31 2007//
 /latin1_to_gsm.h/1.1/Sat Oct  7 15:18:04 2006//
-/list.c/1.45/Sun Jan 14 19:21:31 2007//
-/list.h/1.26/Sun Jan 14 19:21:31 2007//
 /log.h/1.24/Sun Jan 14 19:21:32 2007//
 /md5.c/1.8/Sun Jan 14 19:21:31 2007//
 /md5.h/1.7/Sun Jan 14 19:21:31 2007//
@@ -81,4 +79,6 @@
 /cfg.def/1.128/Wed Mar 21 22:22:14 2007//
 /gwlib.c/1.24/Wed May  9 09:11:24 2007//
 /log.c/1.53/Wed May  9 09:11:17 2007//
+/list.c/1.46/Tue May 15 18:09:17 2007//
+/list.h/1.27/Tue May 15 16:15:00 2007//
 D
=== gwlib/cfg.def
==================================================================
--- gwlib/cfg.def	(revision 288)
+++ gwlib/cfg.def	(local)
@@ -123,6 +123,8 @@
     OCTSTR(sms-outgoing-queue-limit)
     OCTSTR(sms-resend-freq)
     OCTSTR(sms-resend-retry)
+    OCTSTR(sms-combine-concatenated-mo)
+    OCTSTR(sms-combine-concatenated-mo-timeout)
 )