Merge pull request #231 from zoufou/v0.6-beta

v0.6b32

.travis.yml

@@ -4,7 +4,7 @@ python:
 # Command to install dependencies
 install:
     - python setup.py sdist
-    - sudo pip install dist/jasmin-0.6b28.tar.gz
+    - sudo pip install dist/jasmin-0.6b32.tar.gz
     - sudo cp misc/config/init-script/jasmind-ubuntu /etc/init.d/jasmind
     - sudo update-rc.d jasmind defaults
 # Commands to run tests:

jasmin/__init__.py

@@ -5,7 +5,7 @@
 
 MAJOR = 0
 MINOR = 6
-PATCH = 28
+PATCH = 32
 META = 'b'
 
 def get_version():

jasmin/config/tools.py

@@ -49,6 +49,8 @@ def _getint(self, section, option, default=None):
             return default
         if self.config.has_option(section, option) == False:
             return default
+        if self.config.get(section, option) == 'None':
+            return default
 
         return self.config.getint(section, option)
 

jasmin/managers/clients.py

@@ -4,6 +4,7 @@
 import time
 import datetime
 import jasmin
+from logging.handlers import TimedRotatingFileHandler
 from twisted.spread import pb
 from twisted.internet import defer
 from jasmin.protocols.smpp.services import SMPPClientService
@@ -49,7 +50,8 @@ def setConfig(self, SMPPClientPBConfig):
         self.log = logging.getLogger(LOG_CATEGORY)
         if len(self.log.handlers) != 1:
             self.log.setLevel(self.config.log_level)
-            handler = logging.FileHandler(filename=self.config.log_file)
+            handler = TimedRotatingFileHandler(filename=self.config.log_file, 
+                when = self.config.log_rotate)
             formatter = logging.Formatter(self.config.log_format, 
                                           self.config.log_date_format)
             handler.setFormatter(formatter)
@@ -556,7 +558,12 @@ def perspective_submit_sm(self, cid, SubmitSmPDU, priority = 1, validity_period
 
         # Publishing a pickled PDU
         self.log.debug('Publishing SubmitSmPDU with routing_key=%s, priority=%s' % (pubQueueName, priority))
-        c = SubmitSmContent(PickledSubmitSmPDU, responseQueueName, priority, validity_period, submit_sm_resp_bill = submit_sm_resp_bill)
+        c = SubmitSmContent(PickledSubmitSmPDU, 
+            responseQueueName, 
+            priority, 
+            validity_period, 
+            submit_sm_resp_bill = submit_sm_resp_bill,
+            source_connector = 'httpapi' if source_connector == 'httpapi' else 'smppsapi')
         yield self.amqpBroker.publish(exchange='messaging', routing_key=pubQueueName, content=c)
 
         if source_connector == 'httpapi' and dlr_url is not None:

jasmin/managers/configs.py

@@ -40,6 +40,7 @@ def __init__(self, config_file = None):
         self.log_file = self._get('client-management', 
                                   'log_file', 
                                   '/var/log/jasmin/smppclient-manager.log')
+        self.log_rotate = self._get('client-management', 'log_rotate', 'W6')
         self.log_format = self._get('client-management', 
                                     'log_format', 
                                     DEFAULT_LOGFORMAT)
@@ -76,13 +77,18 @@ def __init__(self, config_file = None):
         self.submit_max_age_smppc_not_ready = self._getint('sm-listener', 
             'submit_max_age_smppc_not_ready', 
             1200)
+
+        self.submit_retrial_delay_smppc_not_ready = self._getint('sm-listener', 
+            'submit_retrial_delay_smppc_not_ready', 
+            False)
 
         self.log_level = logging.getLevelName(self._get('sm-listener', 
                                                         'log_level', 
                                                         'INFO'))
         self.log_file = self._get('sm-listener', 
                                   'log_file', 
                                   '/var/log/jasmin/messages.log')
+        self.log_rotate = self._get('sm-listener', 'log_rotate', 'midnight')
         self.log_format = self._get('sm-listener', 
                                     'log_format', 
                                     DEFAULT_LOGFORMAT)

jasmin/managers/content.py

@@ -6,6 +6,7 @@
 import pickle
 import datetime
 from txamqp.content import Content
+from jasmin.protocols.smpp.protocol import SMPPServerProtocol
 
 class InvalidParameterError(Exception):
     """Raised when a parameter is invalid
@@ -95,7 +96,7 @@ def __init__(self, message_status, msgid, system_id, source_addr, destination_ad
 class SubmitSmContent(PDU):
     "A SMPP SubmitSm Content"
 
-    def __init__(self, body, replyto, priority = 1, expiration = None, msgid = None, submit_sm_resp_bill = None):
+    def __init__(self, body, replyto, priority = 1, expiration = None, msgid = None, submit_sm_resp_bill = None, source_connector = 'httpapi'):
         props = {}
 
         # RabbitMQ does not support priority (yet), anyway, we may use any other amqp broker that supports it
@@ -104,14 +105,16 @@ def __init__(self, body, replyto, priority = 1, expiration = None, msgid = None,
         if priority < 0 or priority > 3:
             raise InvalidParameterError("Priority must be set from 0 to 3, it is actually set to %s" % 
                                         priority)
+        if source_connector not in ['httpapi', 'smppsapi']:
+            raise InvalidParameterError('Invalid source_connector value: %s.')
         if msgid is None:
             msgid = randomUniqueId()
 
         props['priority'] = priority
         props['message-id'] = msgid
         props['reply-to'] = replyto
 
-        props['headers'] = {}
+        props['headers'] = {'source_connector': source_connector}
         if submit_sm_resp_bill is not None:
             props['headers']['submit_sm_resp_bill'] = submit_sm_resp_bill
         if expiration is not None:

jasmin/managers/listeners.py

@@ -1,7 +1,9 @@
 #pylint: disable-msg=W0401,W0611
+import re
 import logging
 import pickle
 import struct
+from logging.handlers import TimedRotatingFileHandler
 from datetime import datetime, timedelta
 from dateutil import parser
 from twisted.internet import defer
@@ -18,6 +20,42 @@
 
 LOG_CATEGORY = "jasmin-sm-listener"
 
+def SubmitSmPDUUpdate(fCallback):
+    '''Will extract SubmitSmPDU and update it (if needed) then pass it to fCallback'''
+    def update_submit_sm_pdu(self, *args, **kwargs):
+        message = args[0]
+        SubmitSmPDU = pickle.loads(message.content.body)
+
+        if 'headers' in message.content.properties:
+            headers = message.content.properties['headers']
+            """SubmitSmPDU is sent through httpapi, in this case, some params that cannot be defined
+            through the api must be set here (from the connector config):"""
+            if 'source_connector' in headers and headers['source_connector'] == 'httpapi':
+                update_params = [
+                    'protocol_id', 
+                    'replace_if_present_flag', 
+                    'dest_addr_ton',
+                    'source_addr_npi',
+                    'dest_addr_npi',
+                    'service_type',
+                    'source_addr_ton',
+                    'sm_default_msg_id',
+                ]
+
+                for param in update_params:
+                    _pdu = SubmitSmPDU
+
+                    # Set param in main pdu
+                    _pdu.params[param] = getattr(self.SMPPClientFactory.config, param)
+
+                    # Set param in sub-pdus (multipart use case)
+                    while hasattr(_pdu, 'nextPdu'):
+                        _pdu = _pdu.nextPdu
+                        _pdu.params[param] = getattr(self.SMPPClientFactory.config, param)
+
+        return fCallback(self, message, SubmitSmPDU)
+    return update_submit_sm_pdu
+
 class SMPPClientSMListener:
     debug_it = {'rejectCount': 0}
     '''
@@ -44,7 +82,8 @@ def __init__(self, SMPPClientSMListenerConfig, SMPPClientFactory, amqpBroker, re
         self.log = logging.getLogger(LOG_CATEGORY)
         if len(self.log.handlers) != 1:
             self.log.setLevel(self.config.log_level)
-            handler = logging.FileHandler(filename=self.config.log_file)
+            handler = TimedRotatingFileHandler(filename=self.config.log_file, 
+                when = self.config.log_rotate)
             formatter = logging.Formatter(self.config.log_format, self.config.log_date_format)
             handler.setFormatter(formatter)
             self.log.addHandler(handler)
@@ -110,15 +149,15 @@ def rejectMessage(self, message, requeue = 0):
     @defer.inlineCallbacks
     def ackMessage(self, message):
         yield self.amqpBroker.chan.basic_ack(message.delivery_tag)
-
+
+    @SubmitSmPDUUpdate
     @defer.inlineCallbacks
-    def submit_sm_callback(self, message):
+    def submit_sm_callback(self, message, SubmitSmPDU):
         """This callback is a queue listener
         it is called whenever a message was consumed from queue
         c.f. test_amqp.ConsumeTestCase for use cases
         """
         msgid = message.content.properties['message-id']
-        SubmitSmPDU = pickle.loads(message.content.body)
 
         self.submit_sm_q.get().addCallback(self.submit_sm_callback).addErrback(self.submit_sm_errback)
 
@@ -183,14 +222,19 @@ def submit_sm_callback(self, message):
                 yield self.rejectMessage(message)
                 defer.returnValue(False)
             else:
-                self.log.error("SMPPC [cid:%s] is not connected: Requeuing (#%s) SubmitSmPDU[%s], aged %s seconds." % (
+                if self.config.submit_retrial_delay_smppc_not_ready != False:
+                    delay_str = ' with delay %s seconds' % self.config.submit_retrial_delay_smppc_not_ready
+                else:
+                    delay_str = ''
+                self.log.error("SMPPC [cid:%s] is not connected: Requeuing (#%s) SubmitSmPDU[%s]%s, aged %s seconds." % (
                     self.SMPPClientFactory.config.id, 
                     self.submit_retrials[msgid],
                     msgid,
+                    delay_str,
                     msgAge.seconds,
                     )
                 )
-                yield self.rejectAndRequeueMessage(message)
+                yield self.rejectAndRequeueMessage(message, delay = self.config.submit_retrial_delay_smppc_not_ready)
                 defer.returnValue(False)
         # SMPP Client should be already bound as transceiver or transmitter
         if self.SMPPClientFactory.smpp.isBound() is False:
@@ -207,14 +251,19 @@ def submit_sm_callback(self, message):
                 yield self.rejectMessage(message)
                 defer.returnValue(False)
             else:
-                self.log.error("SMPPC [cid:%s] is not bound: Requeuing (#%s) SubmitSmPDU[%s], aged %s seconds."% (
+                if self.config.submit_retrial_delay_smppc_not_ready != False:
+                    delay_str = ' with delay %s seconds' % self.config.submit_retrial_delay_smppc_not_ready
+                else:
+                    delay_str = ''
+                self.log.error("SMPPC [cid:%s] is not bound: Requeuing (#%s) SubmitSmPDU[%s]%s, aged %s seconds."% (
                     self.SMPPClientFactory.config.id, 
                     self.submit_retrials[msgid],
                     msgid,
+                    delay_str,
                     msgAge,
                     )
                 )
-                yield self.rejectAndRequeueMessage(message)
+                yield self.rejectAndRequeueMessage(message, delay = self.config.submit_retrial_delay_smppc_not_ready)
                 defer.returnValue(False)
 
         self.log.debug("Sending SubmitSmPDU through SMPPClientFactory")
@@ -292,7 +341,7 @@ def submit_sm_resp_event(self, r, amqpMessage):
                                   else amqpMessage.content.properties['headers']['expiration'],
                            r.request.params['source_addr'],
                            r.request.params['destination_addr'],
-                           short_message
+                           re.sub(r'[^\x20-\x7E]+','.', short_message)
                            ))
         else:
             # Message must be retried ?
@@ -319,7 +368,7 @@ def submit_sm_resp_event(self, r, amqpMessage):
                                   else amqpMessage.content.properties['headers']['expiration'],
                            r.request.params['source_addr'],
                            r.request.params['destination_addr'],
-                           r.request.params['short_message']
+                           re.sub(r'[^\x20-\x7E]+','.', r.request.params['short_message'])
                            ))
 
         # It is a final submit_sm_resp !
@@ -560,6 +609,8 @@ def code_dlr_msgid(self, pdu):
     def deliver_sm_event(self, smpp, pdu, concatenated = False):
         """This event is called whenever a deliver_sm pdu is received through a SMPPc
         It will hand the pdu to the router or a dlr thrower (depending if its a DLR or not).
+        
+        Note: this event will catch data_sm pdus as well
         """
 
         pdu.dlr =  self.SMPPOperationFactory.isDeliveryReceipt(pdu)
@@ -612,7 +663,7 @@ def deliver_sm_event(self, smpp, pdu, concatenated = False):
                            pdu.params['validity_period'],
                            pdu.params['source_addr'],
                            pdu.params['destination_addr'],
-                           pdu.params['short_message']
+                           re.sub(r'[^\x20-\x7E]+','.', pdu.params['short_message'])
                            ))
             else:
                 # Long message part received
@@ -739,7 +790,7 @@ def deliver_sm_event(self, smpp, pdu, concatenated = False):
                                 self.log.debug('Removing SMPPs map for msgid[%s]' % submit_sm_queue_id)
                                 yield self.redisClient.delete('smppsmap:%s' % submit_sm_queue_id)
                 else:
-                    self.log.warn('Got a DLR for an unknown message id: %s' % pdu.dlr['id'])
+                    self.log.warn('Got a DLR for an unknown message id: %s (coded:%s)' % (pdu.dlr['id'], _coded_dlr_id))
             else:
                 self.log.warn('DLR for msgid[%s] is not checked, no valid RC were found' % msgid)
 

jasmin/managers/test/test_contents.py

@@ -23,6 +23,7 @@ def test_normal(self):
         self.assertEquals(c['reply-to'], self.replyto)
         self.assertEquals(c['priority'], 1)
         self.assertEquals(c['headers']['expiration'], self.expiration)
+        self.assertEquals(c['headers']['source_connector'], 'httpapi')
         self.assertNotEquals(c['message-id'], None)
         self.assertTrue('created_at' in c['headers'])
 
@@ -46,6 +47,9 @@ def test_unique_messageid(self):
             self.assertEquals(msgIds.count(c['message-id']), 0, "Collision detected at position %s/%s" % (counter, maxCounter))
             msgIds.append(c['message-id'])
 
+    def test_set_incorrect_source_connector(self):
+        self.assertRaises(InvalidParameterError, SubmitSmContent, self.body, self.replyto, source_connector = 'anythingelse')
+
 class SubmitSmRespContentTestCase(ContentTestCase):
     def test_normal_nopickling(self):
         c = SubmitSmRespContent(self.body, 1, prePickle=False)

jasmin/managers/test/test_managers.py

@@ -73,7 +73,7 @@ def setUp(self, authentication = False):
         self.defaultConfig = SMPPClientConfig(id=defaultSMPPClientId,
                                               username='smppclient1',
                                               reconnectOnConnectionFailure=True,
-                                              port=9002
+                                              port=9002,
                                               )
 
     @defer.inlineCallbacks

jasmin/protocols/cli/configs.py

@@ -20,5 +20,6 @@ def __init__(self, config_file = None):
 
         self.log_level = logging.getLevelName(self._get('jcli', 'log_level', 'INFO'))
         self.log_file = self._get('jcli', 'log_file', '/var/log/jasmin/jcli.log')
+        self.log_rotate = self._get('jcli', 'log_rotate', 'W6')
         self.log_format = self._get('jcli', 'log_format', '%(asctime)s %(levelname)-8s %(process)d %(message)s')
         self.log_date_format = self._get('jcli', 'log_date_format', '%Y-%m-%d %H:%M:%S')

jasmin/protocols/cli/factory.py

@@ -1,5 +1,6 @@
 import re
 import logging
+from logging.handlers import TimedRotatingFileHandler
 from twisted.internet import reactor, defer
 from twisted.internet.protocol import ServerFactory
 from jasmin.protocols.cli.jcli import JCliProtocol
@@ -47,7 +48,8 @@ def __init__(self, config, SMPPClientManagerPB, RouterPB, loadConfigProfileWithC
         self.log = logging.getLogger('jcli')
         if len(self.log.handlers) != 1:
             self.log.setLevel(config.log_level)
-            handler = logging.FileHandler(filename=config.log_file)
+            handler = TimedRotatingFileHandler(filename=self.config.log_file, 
+                when = self.config.log_rotate)
             formatter = logging.Formatter(config.log_format, config.log_date_format)
             handler.setFormatter(formatter)
             self.log.addHandler(handler)

jasmin/protocols/cli/smppccm.py

@@ -11,7 +11,7 @@
 
 # A config map between console-configuration keys and SMPPClientConfig keys.
 SMPPClientConfigKeyMap = {'cid': 'id', 'host': 'host', 'port': 'port', 'username': 'username',
-                       'password': 'password', 'systype': 'systemType', 'logfile': 'log_file', 'loglevel': 'log_level',
+                       'password': 'password', 'systype': 'systemType', 'logfile': 'log_file', 'loglevel': 'log_level', 'logrotate': 'log_rotate',
                        'bind_to': 'sessionInitTimerSecs', 'elink_interval': 'enquireLinkTimerSecs', 'trx_to': 'inactivityTimerSecs',
                        'res_to': 'responseTimerSecs', 'con_loss_retry': 'reconnectOnConnectionLoss', 'con_fail_retry': 'reconnectOnConnectionFailure',
                        'con_loss_delay': 'reconnectOnConnectionLossDelay', 'con_fail_delay': 'reconnectOnConnectionFailureDelay',
@@ -26,7 +26,7 @@
 SMPPClientConfigStringKeys = ['systemType', 'username', 'password', 'addressRange']
 
 # When updating a key from RequireRestartKeys, the connector need restart for update to take effect
-RequireRestartKeys = ['host', 'port', 'username', 'password', 'systemType', 'log_file', 'log_level']
+RequireRestartKeys = ['host', 'port', 'username', 'password', 'systemType']
 
 def castOutputToBuiltInType(key, value):
     'Will cast value to the correct type depending on the key'

jasmin/protocols/cli/test/test_jcli.py

@@ -195,7 +195,7 @@ def test_01_persist_all_configurations(self):
                     {'command': 'gid g1'},
                     {'command': 'uid u1'},
                     {'command': 'username fourat'},
-                    {'command': 'password fouratpwd'},
+                    {'command': 'password fpwd'},
                     {'command': 'ok', 'expect': 'Successfully added User'}]
         yield self._test(r'jcli : ', commands)
         # Add HTTP Connector