[PATCH] Re: MO Concatenation

Index: gw/bb_boxc.c
===================================================================
RCS file: /home/cvs/gateway/gw/bb_boxc.c,v
retrieving revision 1.86
diff -u -r1.86 bb_boxc.c
--- gw/bb_boxc.c	3 Nov 2006 14:18:08 -0000	1.86
+++ gw/bb_boxc.c	8 Jan 2007 04:18:36 -0000
@@ -128,6 +128,8 @@
 /* sms_to_smsboxes thread-id */
 static long sms_dequeue_thread;
 
+/* Flag for handling concatenated incoming messages. */
+static int handle_concatenated_mo;
 
 typedef struct _boxc {
     Connection	*conn;
@@ -154,6 +156,8 @@
 static void boxc_sent_push(Boxc*, Msg*);
 static void boxc_sent_pop(Boxc*, Msg*);
 
+static void init_concat_handler(void);
+static void shutdown_concat_handler(void);
 
 /*-------------------------------------------------
  *  receiver thingies
@@ -968,6 +972,8 @@
     smsbox_by_receiver = NULL;
     
     gwlist_remove_producer(flow_threads);
+    /* Stop concat handling */
+    shutdown_concat_handler();
 }
 
 
@@ -1132,6 +1138,12 @@
     gwlist_add_producer(outgoing_sms);
     gwlist_add_producer(smsbox_list);
 
+    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();
+
     smsbox_running = 1;
     
     if ((sms_dequeue_thread = gwthread_create(sms_to_smsboxes, NULL)) == -1)
@@ -1351,6 +1363,7 @@
     boxid = NULL;
 }
 
+static Msg *check_concatenation(Msg *msg);
 
 /*
  * Route the incoming message to one of the following input queues:
@@ -1371,6 +1384,11 @@
     s = r = NULL;
     gw_assert(msg_type(msg) == sms);
 
+    if (handle_concatenated_mo && msg->sms.sms_type == mo) {
+	 if ((msg = check_concatenation(msg)) == NULL)  
+	      return 1; /* i.e. message has been handled. */
+    }
+    
     /* msg_dump(msg, 0); */
 
     /* 
@@ -1537,8 +1555,8 @@
 
         /* debug("bb.sms", 0, "sms_boxc_router: handling message (%p vs %p)",
 	          msg, startmsg); */
+	 ret = route_incoming_to_boxc(msg);
 
-        ret = route_incoming_to_boxc(msg);
         if (ret == 1)
             startmsg = newmsg = NULL;
         else if (ret == -1) {
@@ -1556,3 +1574,188 @@
 
     gwlist_remove_producer(flow_threads);
 }
+
+
+/*--------------------------------
+ * 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. */
+     Msg *parts[1];
+     /* hidden fields! do not put in an array. */
+} ConcatMsg;
+
+static Dict *incoming_concat_msgs;
+static List *incoming_concat_queue;
+static RWLock *concat_rwlock;
+
+static void destroy_concatMsg(void *msg);
+static void init_concat_handler(void)
+{
+     if (incoming_concat_msgs != NULL)
+	  return;
+     incoming_concat_msgs = dict_create(1024, destroy_concatMsg);
+     incoming_concat_queue = gwlist_create();
+     concat_rwlock = gw_rwlock_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);
+     gw_rwlock_destroy(concat_rwlock);
+
+     incoming_concat_msgs = NULL;
+     incoming_concat_queue = NULL;
+     concat_rwlock = 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])
+	       msg_destroy(msg->parts[i]);
+     octstr_destroy(msg->key);
+     gw_free(msg);
+}
+
+/* Checks if message is concatenated. Returns:
+ * -  NULL if message is part of concatenated set and we need to wait for other parts
+ * - non-NULL if it is not, or if we have found all parts and are returning the message.
+ */
+static Msg *check_concatenation(Msg *msg)
+{
+     int l, iel, refnum, pos, c, part, totalparts, i, sixteenbit;
+     Octstr *udh = msg->sms.udhdata, *key;
+     ConcatMsg *cmsg, *x;
+     char xudh[1024];
+     
+     if (incoming_concat_msgs == NULL
+	 || (l = octstr_len(udh)) == 0) /* ... module not initialised or there is no UDH. */
+	  return msg;
+     
+     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 msg;
+     
+     /* 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 msg;
+     }
+
+     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 %d", msg->sms.sender, refnum);          
+     gw_rwlock_wrlock(concat_rwlock);
+     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);
+	  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. Discarding!",
+		  part, refnum, octstr_get_cstr(msg->sms.sender));
+	  msg_destroy(msg); /* if we have, destroy this one.. */
+     } else {
+	  cmsg->num_parts++;
+	  cmsg->parts[part-1] = msg;
+     }
+     
+     if (cmsg->num_parts < cmsg->total_parts) {  /* wait for more parts. */
+	  msg = NULL;
+	  gw_rwlock_unlock(concat_rwlock);     
+	  goto done;
+     }
+
+     /* we have all the parts: Put them together, mod UDH, return message. */
+     msg = cmsg->parts[0];
+     cmsg->parts[0] = NULL; /* remove from array. */
+     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);
+
+     /* 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);
+
+     gw_rwlock_unlock(concat_rwlock);     
+     
+     /* 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:
+
+     /* Remove any pending messages that are too old. */
+     gw_rwlock_wrlock(concat_rwlock);
+     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);
+
+	  gwlist_delete(incoming_concat_queue, 0,1);
+	  dict_put(incoming_concat_msgs, xkey, NULL); /* delete it. */	  
+	  
+	  info(0, "Time-out waiting for concatenated message '%s'. Dropping message parts", 
+	       octstr_get_cstr(xkey));
+	  octstr_destroy(xkey);
+     }
+     gw_rwlock_unlock(concat_rwlock);
+     return msg;
+}
Index: gw/smsc/smsc_at.c
===================================================================
RCS file: /home/cvs/gateway/gw/smsc/smsc_at.c,v
retrieving revision 1.32
diff -u -r1.32 smsc_at.c
--- gw/smsc/smsc_at.c	6 Oct 2006 15:24:39 -0000	1.32
+++ gw/smsc/smsc_at.c	8 Jan 2007 04:18:42 -0000
@@ -1731,7 +1731,7 @@
         pos++;
         if (udhlen + 1 > len)
             goto msg_error;
-        udh = octstr_copy(pdu, pos, udhlen);
+        udh = octstr_copy(pdu, pos-1, udhlen+1);
         pos += udhlen;
         len -= udhlen + 1;
     } else if (len <= 0) /* len < 0 is impossible, but sure is sure */
Index: gwlib/cfg.def
===================================================================
RCS file: /home/cvs/gateway/gwlib/cfg.def,v
retrieving revision 1.123
diff -u -r1.123 cfg.def
--- gwlib/cfg.def	1 Oct 2006 17:10:23 -0000	1.123
+++ gwlib/cfg.def	8 Jan 2007 04:18:43 -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)