CMFActivity: validate using single SQL request per message

product/CMFActivity/Activity/SQLBase.py

@@ -254,15 +254,14 @@ class SQLBase(Queue):
         LOG('SQLBase', INFO, 'Got a lock error, retrying...')
 
   # Validation private methods
-  def getDependentMessageList(self, db, activate_kw, same_queue):
-    q = db.string_literal
+  def getValidationSQL(self, quote, activate_kw, same_queue):
     validate_list = []
     for k, v in activate_kw.iteritems():
       if v is not None:
         try:
           method = getattr(self, '_validate_' + k, None)
           if method:
-            validate_list.append(' AND '.join(method(v, q)))
+            validate_list.append(' AND '.join(method(v, quote)))
         except Exception:
           LOG('CMFActivity', WARNING, 'invalid %s value: %r' % (k, v),
               error=True)
@@ -271,21 +270,12 @@ class SQLBase(Queue):
           same_queue = False
           break
     if validate_list:
-      message_list = []
-      for line in Results(db.query(
-          "SELECT * FROM %s WHERE processing_node > -10 AND (%s) LIMIT %s" % (
-            self.sql_table, ' OR '.join(validate_list),
-            READ_MESSAGE_LIMIT if same_queue else 1), 0)):
-        m = Message.load(line.message,
-                         line=line,
-                         uid=line.uid,
-                         date=line.date,
-                         processing_node=line.processing_node)
-        if not hasattr(m, 'order_validation_text'): # BBB
-          m.order_validation_text = self.getOrderValidationText(m)
-        message_list.append(m)
-      return message_list
-    return ()
+      return ("SELECT '%s' as activity, uid, date, processing_node,"
+              " priority, group_method_id, message FROM %s"
+              " WHERE processing_node > -10 AND (%s) LIMIT %s" % (
+                type(self).__name__, self.sql_table,
+                ' OR '.join(validate_list),
+                READ_MESSAGE_LIMIT if same_queue else 1))
 
   def _validate_after_method_id(self, *args):
     return sqltest_dict['method_id'](*args),

product/CMFActivity/ActivityTool.py

@@ -57,6 +57,7 @@ from Products.ERP5Type.UnrestrictedMethod import PrivilegedUser
 from zope.site.hooks import setSite
 import transaction
 from App.config import getConfiguration
+from Shared.DC.ZRDB.Results import Results
 
 import Products.Localizer.patches
 localizer_lock = Products.Localizer.patches._requests_lock
@@ -1552,10 +1553,27 @@ class ActivityTool (BaseTool):
     def getDependentMessageList(self, message, validating_queue=None):
       activity_kw = message.activity_kw
       db = self.getSQLConnection()
-      return [(activity, m)
-        for activity in activity_dict.itervalues()
-        for m in activity.getDependentMessageList(
-          db, activity_kw, activity is validating_queue)]
+      quote = db.string_literal
+      queries = []
+      for activity in activity_dict.itervalues():
+        q = activity.getValidationSQL(
+          quote, activity_kw, activity is validating_queue)
+        if q:
+          queries.append(q)
+      if queries:
+        message_list = []
+        for line in Results(db.query("(%s)" % ") UNION ALL (".join(queries))):
+          activity = activity_dict[line.activity]
+          m = Message.load(line.message,
+                           line=line,
+                           uid=line.uid,
+                           date=line.date,
+                           processing_node=line.processing_node)
+          if not hasattr(m, 'order_validation_text'): # BBB
+            m.order_validation_text = activity.getOrderValidationText(m)
+          message_list.append((activity, m))
+        return message_list
+      return ()
 
     # Required for tests (time shift)
     def timeShift(self, delay):