Merge 782239d into 7d1aaea

setup.py

@@ -49,7 +49,7 @@
         'txredis',
         'python-smpp>=0.1.5',
         'pytz',
-        'riak>=2.1',
+        'riak>=2.1,<2.6.0',
         'txJSON-RPC==0.3.1',
         'txTwitter>=0.1.4a',
         'treq',

vumi/transports/dmark/dmark_ussd.py

@@ -174,6 +174,20 @@ def handle_raw_inbound_message(self, request_id, request):
                 }
             })
 
+        # NOTE: Make sure we cleanup any lingering HTTP requests
+        #       in case we timeout. Currently any timed out request
+        #       results in a failure being logged, spamming our Sentry
+        #       instance and mailboxes.
+        d = request.notifyFinish()
+        d.addBoth(self.cleanup_request, request_id)
+
+    def cleanup_request(self, error, request_id):
+        if error:
+            self.log.msg('Error cleaning up request %s: %s' % (
+                request_id, error.getErrorMessage()))
+        if request_id in self._requests:
+            self.remove_request(request_id)
+
     @inlineCallbacks
     def handle_outbound_message(self, message):
         self.emit("DmarkUssdTransport consuming %r" % (message,))

vumi/transports/dmark/tests/test_dmark_ussd.py

@@ -8,9 +8,11 @@
 
 from twisted.internet.defer import inlineCallbacks
 from twisted.internet.task import Clock
+from twisted.internet.error import ConnectionDone
 
 from vumi.message import TransportUserMessage
 from vumi.tests.helpers import VumiTestCase
+from vumi.tests.utils import LogCatcher
 from vumi.transports.dmark import DmarkUssdTransport
 from vumi.transports.httprpc.tests.helpers import HttpRpcTransportHelper
 
@@ -299,6 +301,7 @@ def test_status_quick_response(self):
         self.assertEqual(status['component'], 'response')
         self.assertEqual(status['message'], 'Response sent')
         self.assertEqual(status['type'], 'response_sent')
+        self.flushLoggedErrors()
 
     @inlineCallbacks
     def test_status_degraded_slow_response(self):
@@ -430,3 +433,28 @@ def test_status_down_timeout(self):
         self.assertEqual(status['details'], {
             'response_time': self.transport.request_timeout + 0.1,
         })
+
+    @inlineCallbacks
+    def test_notify_finish_requests_cleanup(self):
+        self.tx_helper.mk_request()
+        [msg] = yield self.tx_helper.wait_for_dispatched_inbound(1)
+        [(request_id, request_info)] = self.transport._requests.items()
+        request = request_info['request']
+        # Force a request finish, this should call `notifyFinish()`
+        # which will cleanup the internal _requests registry
+        request.write('')
+        request.finish()
+        self.assertEqual(self.transport._requests, {})
+
+    @inlineCallbacks
+    def test_notify_finish_requests_cleanup_with_errors(self):
+        self.tx_helper.mk_request()
+        [msg] = yield self.tx_helper.wait_for_dispatched_inbound(1)
+        [(request_id, request_info)] = self.transport._requests.items()
+        request = request_info['request']
+        with LogCatcher() as lc:
+            request.connectionLost(ConnectionDone('Connection Done!'))
+        self.assertEqual(
+            lc.messages(),
+            ['Error cleaning up request %s: ' % (request_id,) +
+             'Connection was closed cleanly: Connection Done!.'])

vumi/transports/httprpc/httprpc.py

@@ -317,12 +317,12 @@ def finish_request(self, request_id, data, code=200, headers={}):
                 request.responseHeaders.setRawHeaders(h_name, h_values)
             request.setResponseCode(code)
             request.write(data)
-            request.finish()
             self.set_request_end(request_id)
             self.remove_request(request_id)
             response_id = "%s:%s:%s" % (request.client.host,
                                         request.client.port,
                                         Transport.generate_message_id())
+            request.finish()
             return response_id
 
     # NOTE: This hackery is required so that we know what to_addr a message

vumi/transports/httprpc/tests/test_httprpc.py

@@ -52,6 +52,7 @@ def test_health(self):
         self.assertEqual(json.loads(result), {
             'pending_requests': 0
         })
+        self.flushLoggedErrors()
 
     @inlineCallbacks
     def test_inbound(self):
@@ -63,6 +64,7 @@ def test_inbound(self):
         [ack] = yield self.tx_helper.wait_for_dispatched_events(1)
         self.assertEqual(ack['user_message_id'], rep['message_id'])
         self.assertEqual(ack['sent_message_id'], rep['message_id'])
+        self.flushLoggedErrors()
 
     @inlineCallbacks
     def test_nack(self):