CMFActivity: change behaviour of serialization_tag on SQLQueue to match SQLDict

At most 1 message for a given serialization tag can be validated (cf [28706]).
Respect priority/date/uid when validating only 1 message (cf [34632]).

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@34841 20353a03-c40f-0410-a6d1-a30d3c3de9de

product/CMFActivity/Activity/SQLBase.py

@@ -36,6 +36,11 @@ from Products.CMFActivity.ActiveObject import (
 from Queue import VALIDATION_ERROR_DELAY
 
 
+def sort_message_key(message):
+  # same sort key as in SQL{Dict,Queue}_readMessageList
+  return message.line.priority, message.line.date, message.uid
+
+
 class SQLBase:
   """
     Define a set of common methods for SQL-based storage of activities.

product/CMFActivity/Activity/SQLDict.py

@@ -35,7 +35,7 @@ from ZODB.POSException import ConflictError
 import sys
 from types import ClassType
 #from time import time
-from SQLBase import SQLBase
+from SQLBase import SQLBase, sort_message_key
 from Products.CMFActivity.ActivityRuntimeEnvironment import (
   ActivityRuntimeEnvironment, getTransactionalVariable)
 from zExceptions import ExceptionFormatter
@@ -516,9 +516,6 @@ class SQLDict(RAMDict, SQLBase):
                                         validation_text_dict, now_date=now_date)
 
         if message_dict:
-          def sort_message_key(message):
-            # same sort key as in SQLDict_readMessageList
-            return message.line.priority, message.line.date, message.uid
           message_unique_dict = {}
           serialization_tag_dict = {}
           distributable_uid_set = set()

product/CMFActivity/Activity/SQLQueue.py

@@ -35,8 +35,7 @@ from ZODB.POSException import ConflictError
 from types import ClassType
 import sys
 from time import time
-from sets import ImmutableSet
-from SQLBase import SQLBase
+from SQLBase import SQLBase, sort_message_key
 from Products.CMFActivity.ActivityRuntimeEnvironment import (
   ActivityRuntimeEnvironment, getTransactionalVariable)
 from zExceptions import ExceptionFormatter
@@ -406,13 +405,27 @@ class SQLQueue(RAMQueue, SQLBase):
           message.order_validation_text = self.getOrderValidationText(message)
           self.getExecutableMessageList(activity_tool, message, message_dict,
                                         validation_text_dict, now_date=now_date)
-        distributable_count = len(message_dict)
-        if distributable_count:
-          activity_tool.SQLBase_assignMessage(table=self.sql_table,
-            processing_node=0, uid=[m.uid for m in message_dict.itervalues()])
-          validated_count += distributable_count
-          if validated_count >= MAX_VALIDATED_LIMIT:
-            return
+        if message_dict:
+          distributable_uid_set = set()
+          serialization_tag_dict = {}
+          for message in message_dict.itervalues():
+            serialization_tag = message.activity_kw.get('serialization_tag')
+            if serialization_tag is None:
+              distributable_uid_set.add(message.uid)
+            else:
+              serialization_tag_dict.setdefault(serialization_tag,
+                                                []).append(message)
+          for message_list in serialization_tag_dict.itervalues():
+            # Sort list of messages to validate the message with highest score
+            message_list.sort(key=sort_message_key)
+            distributable_uid_set.add(message_list[0].uid)
+          distributable_count = len(distributable_uid_set)
+          if distributable_count:
+            activity_tool.SQLBase_assignMessage(table=self.sql_table,
+              processing_node=0, uid=tuple(distributable_uid_set))
+            validated_count += distributable_count
+            if validated_count >= MAX_VALIDATED_LIMIT:
+              return
         offset += READ_MESSAGE_LIMIT
 
   # Validation private methods

product/CMFActivity/tests/testCMFActivity.py

@@ -2899,19 +2899,12 @@ class TestCMFActivity(ERP5TypeTestCase, LogInterceptor):
     self.assertEqual(len(result), 2)
     activity_tool.distribute()
     result = activity_tool.getMessageList()
-    # If activity is SQLDict, serialization tag prevents validating the same
-    # serialization tagged messages simultaneously.
-    # If activity is SQLQueue, this does not happen.
-    if activity=='SQLDict':
-      # one is validated.
-      message, = [x for x in result if x.processing_node == 0]
-      self.assertEqual(message.method_id, 'getId')
-      # the other one is still waiting for validation.
-      message, = [x for x in result if x.processing_node == -1]
-      self.assertEqual(message.method_id, 'getTitle')
-    else:
-      # both are validated at once.
-      self.assertEqual(len([x for x in result if x.processing_node == 0]), 2)
+    # at most 1 activity for a given serialization tag can be validated
+    message, = [x for x in result if x.processing_node == 0]
+    self.assertEqual(message.method_id, 'getId')
+    # the other one is still waiting for validation
+    message, = [x for x in result if x.processing_node == -1]
+    self.assertEqual(message.method_id, 'getTitle')
     self.tic()
     result = activity_tool.getMessageList()
     self.assertEqual(len(result), 0)