Merge pull request #1375 from praekelt/feature/GEM-1-conversation-men…

…trics-middleware

Improve conversation middleware

go/vumitools/middleware.py

@@ -5,6 +5,7 @@
 
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.task import LoopingCall
 
 from vumi.config import (
     ConfigBool, ConfigDict, ConfigFloat, ConfigInt, ConfigList, ConfigRiak,
@@ -607,6 +608,9 @@ class ConversationMetricsMiddlewareConfig(BaseMiddleware.CONFIG_CLASS):
 
     redis_manager = ConfigDict(
         "Redis configuration parameters", default={}, static=True)
+    local_store_reset_timer = ConfigInt(
+        "Time in seconds to store conversations keys for locally.",
+        default=300, static=True)
 
 
 class ConversationMetricsMiddleware(BaseMiddleware):
@@ -631,10 +635,18 @@ def setup_middleware(self):
         self.redis_manager = yield TxRedisManager.from_config(
             self.config.redis_manager)
         self.redis = self.redis_manager.sub_manager(self.SUBMANAGER_PREFIX)
+        self.local_recent_convs = set()
+        self._looper = LoopingCall(self.reset_local_recent_convs)
+        self._looper.start(self.config.local_store_reset_timer)
 
     def teardown_middleware(self):
+        if self._looper.running:
+            self._looper.stop()
         return self.redis_manager.close_manager()
 
+    def reset_local_recent_convs(self):
+        self.local_recent_convs.clear()
+
     def record_conv_seen(self, msg):
         mdh = MessageMetadataHelper(None, msg)
         conv_key = mdh.get_conversation_key()
@@ -645,9 +657,11 @@ def record_conv_seen(self, msg):
         conv_details = '{"account_key": "%s","conv_key": "%s"}' % \
             (acc_key, conv_key)
 
-        # Note: This set will be emptied by a celery task that publishes the
-        # metrics for conversations we have seen
-        return self.redis.sadd(self.RECENT_CONV_KEY, conv_details)
+        if conv_details not in self.local_recent_convs:
+            self.local_recent_convs.add(conv_details)
+            # Note: This set will be emptied by a celery task that publishes
+            # the metrics for conversations we have seen
+            self.redis.sadd(self.RECENT_CONV_KEY, *self.local_recent_convs)
 
     @inlineCallbacks
     def handle_inbound(self, message, connector_name):

go/vumitools/tests/test_middleware.py

@@ -1029,36 +1029,47 @@ def setUp(self):
             u'bulk_message', name=u'Test Conversation', started=True)
 
     @inlineCallbacks
-    def assert_conv_in_redis(self, mw, msg):
+    def assert_conv_key_stored(self, mw, msg):
         value = yield mw.redis.smembers(
             ConversationMetricsMiddleware.RECENT_CONV_KEY)
         conv_details = '{"account_key": "%s","conv_key": "%s"}' % \
             (self.conv.user_account.key, self.conv.key)
         self.assertTrue(conv_details in value)
         self.assertEqual(len(value), 1)
+        self.assertIn(conv_details, mw.local_recent_convs)
 
     @inlineCallbacks
-    def assert_conv_not_in_redis(self, mw):
+    def assert_conv_key_not_stored(self, mw):
         value = yield mw.redis.smembers(
             ConversationMetricsMiddleware.RECENT_CONV_KEY)
         self.assertSetEqual(value, set([]))
+        self.assertSetEqual(mw.local_recent_convs, set([]))
 
     @inlineCallbacks
     def test_inbound_message(self):
         mw = yield self.mw_helper.create_middleware()
         msg_helper = GoMessageHelper(vumi_helper=self.mw_helper)
+        yield self.assert_conv_key_not_stored(mw)
 
-        yield self.assert_conv_not_in_redis(mw)
         [msg] = yield msg_helper.add_inbound_to_conv(self.conv, 1)
         yield mw.handle_inbound(msg, "conn_1")
-        yield self.assert_conv_in_redis(mw, msg)
+        yield self.assert_conv_key_stored(mw, msg)
 
     @inlineCallbacks
     def test_outbound_message(self):
         mw = yield self.mw_helper.create_middleware()
         msg_helper = GoMessageHelper(vumi_helper=self.mw_helper)
+        yield self.assert_conv_key_not_stored(mw)
 
-        yield self.assert_conv_not_in_redis(mw)
         [msg] = yield msg_helper.add_outbound_to_conv(self.conv, 1)
         yield mw.handle_outbound(msg, "conn_1")
-        yield self.assert_conv_in_redis(mw, msg)
+        yield self.assert_conv_key_stored(mw, msg)
+
+    @inlineCallbacks
+    def test_reset_local_recent_convs(self):
+        mw = yield self.mw_helper.create_middleware()
+        mw.local_recent_convs.update(["conv1", "conv2"])
+
+        mw.reset_local_recent_convs()
+
+        self.assertEqual(mw.local_recent_convs, set([]))