Add support for multiple after methods and broadcast.

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

product/CMFActivity/Activity/SQLQueue.py

@@ -59,6 +59,7 @@ class SQLQueue(RAMQueue):
       activity_tool.SQLQueue_writeMessage(path = '/'.join(m.object_path) ,
                                           method_id = m.method_id,
                                           priority = m.activity_kw.get('priority', 1),
+                                          broadcast = m.activity_kw.get('broadcast', 0),
                                           message = self.dumpMessage(m),
                                           date = m.activity_kw.get('at_date', DateTime()))
 
@@ -66,7 +67,7 @@ class SQLQueue(RAMQueue):
     # Erase all messages in a single transaction
     LOG("prepareDeleteMessage", 0, str(m.__dict__))
     activity_tool.SQLQueue_delMessage(uid = m.uid)
-    
+
   def dequeueMessage(self, activity_tool, processing_node):
     if hasattr(activity_tool,'SQLQueue_readMessageList'):
       now_date = DateTime()
@@ -90,7 +91,7 @@ class SQLQueue(RAMQueue):
         # Make sure object exists
         validation_state = m.validate(self, activity_tool)
         if validation_state is not VALID:
-          if validation_state in (EXCEPTION, INVALID_PATH):          
+          if validation_state in (EXCEPTION, INVALID_PATH):
             if line.priority > MAX_PRIORITY:
               # This is an error
               activity_tool.SQLQueue_assignMessage(uid=line.uid, processing_node = VALIDATE_ERROR_STATE)
@@ -212,32 +213,49 @@ class SQLQueue(RAMQueue):
       #LOG('distribute count',0,str(len(result)) )
       #LOG('distribute count',0,str(map(lambda x:x.uid, result)))
       #get_transaction().commit() # Release locks before starting a potentially long calculation
-      uid_list = map(lambda x:x.uid, result)[0:100]
-      for uid in uid_list:
-        #LOG("distribute", 0, "assign %s" % uid)
-        activity_tool.SQLQueue_assignMessage(uid=uid, processing_node=processing_node)
-        #get_transaction().commit() # Release locks immediately to allow processing of messages
-        processing_node = processing_node + 1
-        if processing_node > node_count:
-          processing_node = 1 # Round robin
+      result = list(result)[0:100]
+      for line in result:
+        broadcast = line.broadcast
+        uid = line.uid
+        if broadcast:
+          # Broadcast messages must be distributed into all nodes.
+          activity_tool.SQLQueue_assignMessage(processing_node=1, uid=uid)
+          for node in range(2, node_count+1):
+            activity_tool.SQLQueue_writeMessage( path = line.path,
+                                                method_id = line.method_id,
+                                                priority = line.priority,
+                                                broadcast = 1,
+                                                processing_node = node,
+                                                message = line.message,
+                                                date = line.date)
+        else:
+          #LOG("distribute", 0, "assign %s" % uid)
+          activity_tool.SQLQueue_assignMessage(uid=uid, processing_node=processing_node)
+          #get_transaction().commit() # Release locks immediately to allow processing of messages
+          processing_node = processing_node + 1
+          if processing_node > node_count:
+            processing_node = 1 # Round robin
 
   # Validation private methods
   def _validate_after_method_id(self, activity_tool, message, value):
     # Count number of occurances of method_id
-    LOG('SQLQueue._validate_after_method_id, message',0,message)
-    LOG('SQLQueue._validate_after_method_id, value',0,value)
+    #get_transaction().commit()
+    if type(value) == type(''):
+      value = [value]
     result = activity_tool.SQLQueue_validateMessageList(method_id=value, message_uid=None, path=None)
+    LOG('SQLQueue._validate_after_method_id, method_id',0,value)
+    LOG('SQLQueue._validate_after_method_id, result[0].uid_count',0,result[0].uid_count)
     if result[0].uid_count > 0:
       return INVALID_ORDER
     return VALID
-            
+
   def _validate_after_path(self, activity_tool, message, value):
     # Count number of occurances of path
     result = activity_tool.SQLQueue_validateMessageList(method_id=None, message_uid=None, path=value)
     if result[0].uid_count > 0:
       return INVALID_ORDER
     return VALID
-            
+
   def _validate_after_message_uid(self, activity_tool, message, value):
     # Count number of occurances of message_uid
     result = activity_tool.SQLQueue_validateMessageList(method_id=None, message_uid=value, path=None)
@@ -245,8 +263,8 @@ class SQLQueue(RAMQueue):
       return INVALID_ORDER
     return VALID
 
-  # Required for tests (time shift)        
-  def timeShift(self, activity_tool, delay):    
+  # Required for tests (time shift)
+  def timeShift(self, activity_tool, delay):
     """
       To simulate timeShift, we simply substract delay from
       all dates in SQLDict message table

product/CMFActivity/skins/activity/SQLQueue_assignMessage.zsql

@@ -10,6 +10,7 @@ class_file:
 <params>path
 processing_node
 method_id
+broadcast
 uid:int=0</params>
 UPDATE message_queue
 SET
@@ -19,3 +20,7 @@ WHERE
 <dtml-if path> path = <dtml-sqlvar path type="string"> 
 <dtml-else> uid = <dtml-sqlvar uid type="int"> </dtml-if>
 <dtml-if method_id> AND method_id = <dtml-sqlvar method_id type="string"></dtml-if>
+<dtml-if broadcast>
+  AND broadcast = <dtml-sqlvar broadcast type="int">
+</dtml-if>
+

product/CMFActivity/skins/activity/SQLQueue_createMessageTable.zsql

@@ -17,6 +17,7 @@ CREATE TABLE `message_queue` (
   `processing` INT DEFAULT 0,
   `processing_date` datetime,
   `priority` INT DEFAULT 0,
+  `broadcast` INT DEFAULT 0,
   `message` BLOB,
   PRIMARY KEY  (`uid`),
   KEY `date` (`date`),

product/CMFActivity/skins/activity/SQLQueue_validateMessageList.zsql

@@ -17,6 +17,12 @@ FROM
     message_queue
 WHERE
     processing_node >= -1
-<dtml-if method_id>AND method_id = <dtml-sqlvar method_id type="string"> </dtml-if>
+<dtml-if method_id>
+    AND (
+<dtml-in method_id>
+        method_id = <dtml-sqlvar sequence-item type="string"><dtml-if sequence-end><dtml-else> OR </dtml-if>
+</dtml-in>
+    )
+</dtml-if>
 <dtml-if message_uid>AND uid = <dtml-sqlvar message_uid type="int"> </dtml-if>
 <dtml-if path>AND path = <dtml-sqlvar path type="string"></dtml-if>

product/CMFActivity/skins/activity/SQLQueue_writeMessage.zsql

@@ -11,13 +11,16 @@ class_file:
 method_id
 message
 priority
+broadcast
+processing_node=-1
 date</params>
 INSERT INTO message_queue
 SET
 	path = <dtml-sqlvar path type="string">,
   <dtml-if date>date = <dtml-sqlvar date type="string">, <dtml-else>date = <dtml-sqlvar "_.DateTime()" type="string">, </dtml-if> 
 	method_id = <dtml-sqlvar method_id type="string">,
-	processing_node = -1,
+	processing_node = <dtml-sqlvar processing_node type="int">,
+	broadcast = <dtml-sqlvar broadcast type="int">,
 	processing = -1,
 	priority = <dtml-sqlvar priority type="int">,
 	message = <dtml-sqlvar message type="string">