Re: [PATCH] Re: MO Concatenation

Index: doc/userguide/userguide.xml
===================================================================
RCS file: /home/cvs/gateway/doc/userguide/userguide.xml,v
retrieving revision 1.327
diff -u -r1.327 userguide.xml
--- doc/userguide/userguide.xml	22 Jan 2007 15:12:42 -0000	1.327
+++ doc/userguide/userguide.xml	12 Apr 2007 13:28:20 -0000
@@ -1526,6 +1526,14 @@
         </entry>   
      </row>
 
+     <row><entry><literal>smsbox-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>
+
   </tbody>
   </tgroup>
  </table>
Index: gw/bb_smscconn.c
===================================================================
RCS file: /home/cvs/gateway/gw/bb_smscconn.c,v
retrieving revision 1.91
diff -u -r1.91 bb_smscconn.c
--- gw/bb_smscconn.c	31 Jan 2007 12:06:00 -0000	1.91
+++ gw/bb_smscconn.c	12 Apr 2007 13:28:27 -0000
@@ -130,11 +130,20 @@
  */
 Counter *split_msg_counter;
 
+/* Flag for handling concatenated incoming messages. */
+static int handle_concatenated_mo;
+/* 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
@@ -396,6 +405,25 @@
      * Scope: internal routing (to smsc-ids)
      */
     if ((rc = route_incoming_to_smsc(conn, copy)) == -1) {
+        int ret;
+        /* Before routing to bearerbox, do concat handling
+         * and replace copy as such.
+         */
+        if (handle_concatenated_mo && copy->sms.sms_type == mo &&
+            (ret = check_concatenation(&copy, conn->id)) != concat_none)
+            if (ret == concat_pending) {
+                counter_increase(incoming_sms_counter); /* ?? */
+                if (conn != NULL) counter_increase(conn->received);		   
+                msg_destroy(sms);
+                return SMSCCONN_SUCCESS;
+            }  else if (ret == concat_complete) { /* Combined sms received! save new one since it is now combined. */ 
+                msg_destroy(sms);
+                sms = msg_duplicate(copy); /* Change the sms. */
+            } else if (ret == concat_error) {
+                /* failed to save, go away. */
+                msg_destroy(copy);
+                return SMSCCONN_FAILED_TEMPORARILY;
+            } 	 
         /*
          * Now try to route the message to a specific smsbox
          * connection based on the existing msg->sms.boxc_id or
@@ -563,6 +591,12 @@
     else
         info(0, "SMS resend retry set to %ld.", sms_resend_retry);
 
+    if (cfg_get_bool(&handle_concatenated_mo, grp, octstr_imm("smsbox-combine-concatenated-mo")) == -1)
+        handle_concatenated_mo = 1; /* default is TRUE. */
+
+    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) {
@@ -806,6 +840,9 @@
     counter_destroy(split_msg_counter);
     gw_rwlock_destroy(&smsc_list_lock);
 
+    /* Stop concat handling */
+    shutdown_concat_handler();
+
     smsc_running = 0;
 }
 
@@ -1018,6 +1055,8 @@
 	return (smsc2_rout(msg));	/* re-try */
 
     msg_destroy(msg);
+    if (handle_concatenated_mo)
+        clear_old_concat_parts(); /* clear old concat parts. */
     return 1;
 }
 
@@ -1078,3 +1117,219 @@
     return -1; 
 }
 
+
+/*--------------------------------
+ * incoming concatenated messages handling
+ */
+
+#define CONCATENATED_TIMEOUT 1800
+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. */
+    Msg *parts[1];
+    /* hidden fields! do not put in an array. */
+} ConcatMsg;
+
+static Dict *incoming_concat_msgs;
+static List *incoming_concat_queue;
+static Mutex *concat_lock;
+
+static void destroy_concatMsg(void *msg);
+static void init_concat_handler(void)
+{
+    extern max_incoming_sms_qlength;
+
+    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);
+    incoming_concat_queue = gwlist_create();
+    concat_lock = mutex_create();
+    debug("bb",0,"smsbox MO concatenated message handling enabled");
+}
+
+static void shutdown_concat_handler(void)
+{
+    if (incoming_concat_msgs == NULL)
+        return;
+    dict_destroy(incoming_concat_msgs);
+    gwlist_destroy(incoming_concat_queue,NULL);
+    mutex_destroy(concat_lock);
+
+    incoming_concat_msgs = NULL;
+    incoming_concat_queue = NULL;
+    concat_lock = NULL;
+    debug("bb",0,"smsbox MO concatenated message handling cleaned up");
+}
+
+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]);
+        }
+    octstr_destroy(msg->key);
+    gw_free(msg);
+}
+
+static void clear_old_concat_parts(void)
+{
+
+    ConcatMsg *x;
+    if (incoming_concat_msgs != NULL) /* not initialised, go away */
+        return;    
+    /* Remove any pending messages that are too old. */
+    mutex_lock(concat_lock);
+    while (gwlist_len(incoming_concat_queue) > 0 && 
+           (x = gwlist_get(incoming_concat_queue, 0)) != NULL &&
+           time(NULL)  - x->trecv > CONCATENATED_TIMEOUT) {
+        Octstr *xkey = octstr_duplicate(x->key);
+
+        x->ack = ack_failed;
+        gwlist_delete(incoming_concat_queue, 0,1);
+        dict_remove(incoming_concat_msgs, xkey); /* delete it. */	  
+	  
+        warning(0, "Time-out waiting for concatenated message '%s'. Dropping message parts", 
+                octstr_get_cstr(xkey));
+        octstr_destroy(xkey);
+    }
+    mutex_unlock(concat_lock); 
+}
+
+/* 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, *x;
+    int ret = concat_complete;
+
+    if (incoming_concat_msgs == NULL
+        || (l = octstr_len(udh)) == 0) /* ... module not initialised or there is no UDH. */
+        return concat_none;
+     
+    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)
+            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("concat.sms", 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 %d", msg->sms.sender, smscid, refnum);          
+    mutex_lock(concat_lock);
+    if ((cmsg = dict_get(incoming_concat_msgs, key)) == NULL) {
+        cmsg = gw_malloc(totalparts * sizeof (cmsg->parts[0]) + sizeof *cmsg);
+        cmsg->refnum = refnum;
+        cmsg->total_parts = totalparts;
+        cmsg->num_parts = 0;
+        cmsg->trecv = time(NULL);
+        cmsg->key = octstr_duplicate(key);
+        cmsg->ack = ack_success;
+        memset(cmsg->parts, 0, (cmsg->total_parts+1)*sizeof cmsg->parts[0]); /* clear it. */	  
+	  
+        dict_put(incoming_concat_msgs, key, cmsg);
+        gwlist_append(incoming_concat_queue, cmsg);
+    }
+    octstr_destroy(key); 
+     
+    /* check if we have seen message part before... */
+    if (cmsg->parts[part-1] != NULL) {	  
+        warning(0, "Duplicate message part %d, ref %d, from %s. Discarded!",
+                part, refnum, octstr_get_cstr(msg->sms.sender));
+        store_save_ack(msg, ack_success);
+        msg_destroy(msg); 
+        *pmsg = msg = NULL;
+    } else {
+        cmsg->num_parts++;
+        cmsg->parts[part-1] = msg;
+    }
+     
+    if (cmsg->num_parts < cmsg->total_parts) {  /* wait for more parts. */
+        *pmsg = msg = NULL;
+        mutex_unlock(concat_lock);     
+        ret = concat_pending;
+        goto done;
+    }
+
+    /* 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("concat.sms",0,"Received all concatenated message parts from %s, refnum %d",
+          octstr_get_cstr(msg->sms.sender), 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) {	  
+        msg_destroy(msg);
+        *pmsg = msg = NULL;
+        ret = concat_error;
+        mutex_unlock(concat_lock);     
+        goto done;
+    } else 
+        *pmsg = msg; /* return the message part. */
+
+    /* Delete it from the queue and from the Dict. */
+    gwlist_delete_equal(incoming_concat_queue, cmsg);
+    key = octstr_duplicate(cmsg->key); 
+    dict_put(incoming_concat_msgs, key, NULL); /* delete it. */
+    octstr_destroy(key);
+
+    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("concat.sms", 0, "Got full message [ref %d] of message from %s. Dumping: ",
+          refnum, octstr_get_cstr(msg->sms.sender));
+    msg_dump(msg,0);
+
+  done:
+    return ret;
+}
Index: gwlib/cfg.def
===================================================================
RCS file: /home/cvs/gateway/gwlib/cfg.def,v
retrieving revision 1.127
diff -u -r1.127 cfg.def
--- gwlib/cfg.def	27 Feb 2007 16:33:44 -0000	1.127
+++ gwlib/cfg.def	12 Apr 2007 13:28:30 -0000
@@ -85,6 +85,7 @@
     OCTSTR(smsbox-port)
     OCTSTR(smsbox-port-ssl)
     OCTSTR(smsbox-max-pending)
+    OCTSTR(smsbox-combine-concatenated-mo)
     OCTSTR(wapbox-port)
     OCTSTR(wapbox-port-ssl)
     OCTSTR(box-deny-ip)