Re: [PATCH] Re: MO Concatenation

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	26 Feb 2007 17:53:20 -0000
@@ -130,11 +130,19 @@
  */
 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);
 
 /*---------------------------------------------------------------------------
  * CALLBACK FUNCTIONS
@@ -396,6 +404,26 @@
      * 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) {
+		   bb_alog_sms(conn, sms, "CONCATENATED SMS Part Received, stored");	     		   
+		   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;
 }
 
@@ -1078,3 +1115,207 @@
     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 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]) {
+	       store_save_ack(msg->parts[i], msg->ack);
+	       msg_destroy(msg->parts[i]);
+	  }
+     octstr_destroy(msg->key);
+     gw_free(msg);
+}
+
+/* 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);          
+     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);
+	  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) {	  
+	  debug("concat.sms",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;
+	  gw_rwlock_unlock(concat_rwlock);     
+	  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;
+	  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);
+
+     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);
+
+	  x->ack = ack_failed;
+	  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 ret;
+}
Index: gwlib/cfg.def
===================================================================
RCS file: /home/cvs/gateway/gwlib/cfg.def,v
retrieving revision 1.126
diff -u -r1.126 cfg.def
--- gwlib/cfg.def	22 Jan 2007 15:12:42 -0000	1.126
+++ gwlib/cfg.def	26 Feb 2007 17:53:26 -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)

Index: gw/smsc/smsc_at.c
===================================================================
RCS file: /home/cvs/gateway/gw/smsc/smsc_at.c,v
retrieving revision 1.33
diff -u -r1.33 smsc_at.c
--- gw/smsc/smsc_at.c	7 Jan 2007 23:52:53 -0000	1.33
+++ gw/smsc/smsc_at.c	26 Feb 2007 17:53:26 -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 */