SMPP needs to grow support for max TPS outbound.

vumi/transports/smpp/config.py

@@ -96,3 +96,9 @@ class SmppTransportConfig(Transport.CONFIG_CLASS):
         "delay before reconnecting. In these cases a 45s "
         "`initial_reconnect_delay` is recommended. Default 55.",
         default=55, static=True)
+    mt_tps = ConfigInt(
+        'Mobile Terminated Transactions per Second. The Maximum Vumi '
+        'messages per second to attempt to put on the wire. '
+        'Defaults to 0 which means no throttling is applied. '
+        '(NOTE: 1 Vumi message may result in multiple PDUs)',
+        default=0, static=True, required=False)

vumi/transports/smpp/smpp_transport.py

@@ -7,6 +7,7 @@
 from twisted.internet.defer import (
     inlineCallbacks, DeferredQueue, maybeDeferred, returnValue, Deferred,
     succeed)
+from twisted.internet.task import LoopingCall
 
 from vumi.reconnecting_client import ReconnectingClientService
 from vumi.transports.base import Transport
@@ -143,12 +144,22 @@ def setup_transport(self):
             self, config.deliver_short_message_processor_config)
         self.submit_sm_processor = config.submit_short_message_processor(
             self, config.submit_short_message_processor_config)
+
         self.sequence_generator = RedisSequence(self.redis)
         self.throttled = None
         self.factory = self.factory_class(self)
 
         self.service = self.start_service(self.factory)
 
+        self.tps_counter = 0
+        self.tps_limit = config.mt_tps
+        if config.mt_tps > 0:
+            self.mt_tps_lc = LoopingCall(self.reset_mt_tps)
+            self.mt_tps_lc.clock = self.clock
+            self.mt_tps_lc.start(1, now=True)
+        else:
+            self.mt_tps_lc = None
+
     def start_service(self, factory):
         config = self.get_static_config()
         service = self.service_class(config.twisted_endpoint, factory)
@@ -159,10 +170,37 @@ def start_service(self, factory):
     def teardown_transport(self):
         if self.service:
             yield self.service.stopService()
+        if self.mt_tps_lc and self.mt_tps_lc.running:
+            self.mt_tps_lc.stop()
         yield self.redis._close()
 
+    def reset_mt_tps(self):
+        if self.throttled and self.need_mt_throttling():
+            self.reset_mt_throttle_counter()
+            self.stop_throttling()
+
+    def reset_mt_throttle_counter(self):
+        self.tps_counter = 0
+
+    def incr_mt_throttle_counter(self):
+        self.tps_counter += 1
+
+    def need_mt_throttling(self):
+        return self.tps_counter > self.tps_limit
+
+    def bind_requires_throttling(self):
+        config = self.get_static_config()
+        return config.mt_tps > 0
+
+    def check_mt_throttling(self):
+        self.incr_mt_throttle_counter()
+        if self.need_mt_throttling():
+            self.start_throttling()
+
     @inlineCallbacks
     def handle_outbound_message(self, message):
+        if self.bind_requires_throttling():
+            self.check_mt_throttling()
         protocol = yield self.service.get_protocol()
         seq_nrs = yield self.submit_sm_processor.handle_outbound_message(
             message, protocol)

vumi/transports/smpp/tests/test_smpp_transport.py

@@ -476,6 +476,26 @@ def test_mt_sms_throttled(self):
         self.assertEqual(event['event_type'], 'ack')
         self.assertEqual(event['user_message_id'], msg['message_id'])
 
+    @inlineCallbacks
+    def test_mt_sms_tps_limits(self):
+        smpp_helper = yield self.get_smpp_helper(config={
+            'mt_tps': 1,
+        })
+        transport = smpp_helper.transport
+
+        msg1 = yield self.tx_helper.make_dispatch_outbound('hello world 1')
+        msg2 = yield self.tx_helper.make_dispatch_outbound('hello world 2')
+
+        self.assertTrue(transport.throttled)
+        [submit_sm_pdu1] = yield smpp_helper.wait_for_pdus(1)
+        self.assertEqual(short_message(submit_sm_pdu1), 'hello world 1')
+
+        self.clock.advance(1)
+
+        self.assertFalse(transport.throttled)
+        [submit_sm_pdu2] = yield smpp_helper.wait_for_pdus(1)
+        self.assertEqual(short_message(submit_sm_pdu2), 'hello world 2')
+
     @inlineCallbacks
     def test_mt_sms_queue_full(self):
         smpp_helper = yield self.get_smpp_helper()