pgq.Consumer: 'table_filter' - filter for ev_extra1

diff --git a/python/pgq/consumer.py b/python/pgq/consumer.py
index 76470b724eb149624d45cecc38f85e690d9ea4bf..77c018c79b31eb4cd74a9769cd636dfdf2c9e987 100644 (file)
--- a/python/pgq/consumer.py
+++ b/python/pgq/consumer.py
@@ -111,6 +111,9 @@ class Consumer(skytools.DBScript):
         # override consumer name
         #consumer_name = %(job_name)s
 
+        # filter out only events for specific tables
+        #table_filter = table1, table2
+
         # whether to use cursor to fetch events (0 disables)
         #pgq_lazy_fetch = 300
 
@@ -198,6 +201,14 @@ class Consumer(skytools.DBScript):
         self.pgq_min_interval = self.cf.get("pgq_batch_collect_interval", '') or None
         self.pgq_min_lag = self.cf.get("pgq_keep_lag", '') or None
 
+        # filter out specific tables only
+        tfilt = []
+        for t in self.cf.getlist('table_filter', ''):
+            tfilt.append(skytools.quote_literal(skytools.fq_name(t)))
+        if len(tfilt) > 0:
+            expr = "ev_extra1 in (%s)" % ','.join(tfilt)
+            self.consumer_filter = expr
+
     def startup(self):
         """Handle commands here.  __init__ does not have error logging."""
         if self.options.register: