ActivityTool.py 44.6 KB
Newer Older
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1 2 3
##############################################################################
#
# Copyright (c) 2002 Nexedi SARL and Contributors. All Rights Reserved.
Jean-Paul Smets's avatar
Jean-Paul Smets committed
4
#                    Jean-Paul Smets-Solanes <jp@nexedi.com>
Jean-Paul Smets's avatar
Jean-Paul Smets committed
5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
#
##############################################################################

29 30 31 32
import socket
import urllib
import threading
import sys
Vincent Pelletier's avatar
Vincent Pelletier committed
33
from types import StringType
34 35
import re

Jean-Paul Smets's avatar
Jean-Paul Smets committed
36
from Products.CMFCore import CMFCorePermissions
Jean-Paul Smets's avatar
Jean-Paul Smets committed
37
from Products.ERP5Type.Core.Folder import Folder
38
from Products.CMFActivity.ActiveResult import ActiveResult
39
from Products.PythonScripts.Utility import allow_class
40
from AccessControl import ClassSecurityInfo, Permissions
Jérome Perrin's avatar
Jérome Perrin committed
41 42 43 44
from AccessControl.SecurityManagement import newSecurityManager
from AccessControl.SecurityManagement import noSecurityManager
from AccessControl.SecurityManagement import setSecurityManager
from AccessControl.SecurityManagement import getSecurityManager
45 46
from Products.CMFCore.utils import UniqueObject, _getAuthenticatedUser, getToolByName
from Globals import InitializeClass, DTMLFile
Jean-Paul Smets's avatar
Jean-Paul Smets committed
47
from Acquisition import aq_base
48
from Acquisition import aq_inner
49
from ActivityBuffer import ActivityBuffer
50
from zExceptions import ExceptionFormatter
51
from BTrees.OIBTree import OIBTree
52

53
from ZODB.POSException import ConflictError
54
from Products.MailHost.MailHost import MailHostError
Jean-Paul Smets's avatar
Jean-Paul Smets committed
55

56
from zLOG import LOG, INFO, WARNING, ERROR
57
from warnings import warn
58 59

try:
60
  from Products.TimerService import getTimerService
61
except ImportError:
62 63
  def getTimerService(self):
    pass
Jean-Paul Smets's avatar
Jean-Paul Smets committed
64

65
# minimal IP:Port regexp
66
NODE_RE = re.compile('^\d+\.\d+\.\d+\.\d+:\d+$')
67

Jean-Paul Smets's avatar
Jean-Paul Smets committed
68 69 70 71
# Using a RAM property (not a property of an instance) allows
# to prevent from storing a state in the ZODB (and allows to restart...)
active_threads = 0
max_active_threads = 1 # 2 will cause more bug to appear (he he)
Vincent Pelletier's avatar
Vincent Pelletier committed
72
is_initialized = False
73 74
tic_lock = threading.Lock() # A RAM based lock to prevent too many concurrent tic() calls
timerservice_lock = threading.Lock() # A RAM based lock to prevent TimerService spamming when busy
75
first_run = True
76 77 78
currentNode = None
ROLE_IDLE = 0
ROLE_PROCESSING = 1
Jean-Paul Smets's avatar
Jean-Paul Smets committed
79 80 81 82

# Activity Registration
activity_dict = {}

83
logging = False
84 85 86 87 88 89 90 91 92

def enableLogging():
  global logging
  logging = True

def disableLogging():
  global logging
  logging = False

93 94 95 96 97 98 99
# Here go ActivityBuffer instances
# Structure:
#  global_activity_buffer[activity_tool_path][thread_id] = ActivityBuffer
global_activity_buffer = {}
from thread import get_ident, allocate_lock
global_activity_buffer_lock = allocate_lock()

Jean-Paul Smets's avatar
Jean-Paul Smets committed
100 101 102
def registerActivity(activity):
  # Must be rewritten to register
  # class and create instance for each activity
103
  #LOG('Init Activity', 0, str(activity.__name__))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
104 105 106 107
  activity_instance = activity()
  activity_dict[activity.__name__] = activity_instance

class Message:
108
  """Activity Message Class.
109

110 111
  Message instances are stored in an activity queue, inside the Activity Tool.
  """
112 113
  def __init__(self, obj, active_process, activity_kw, method_id, args, kw):
    if isinstance(obj, str):
114
      self.object_path = tuple(obj.split('/'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
115
    else:
116
      self.object_path = obj.getPhysicalPath()
Yoshinori Okuji's avatar
Yoshinori Okuji committed
117
    if type(active_process) is StringType:
118 119 120 121 122
      self.active_process = active_process.split('/')
    elif active_process is None:
      self.active_process = None
    else:
      self.active_process = active_process.getPhysicalPath()
123
      self.active_process_uid = active_process.getUid()
124 125 126
    if activity_kw.get('serialization_tag', False) is None:
      # Remove serialization_tag if it's None.
      del activity_kw['serialization_tag']
Jean-Paul Smets's avatar
Jean-Paul Smets committed
127 128 129 130
    self.activity_kw = activity_kw
    self.method_id = method_id
    self.args = args
    self.kw = kw
Jean-Paul Smets's avatar
Jean-Paul Smets committed
131
    self.is_executed = 0
Vincent Pelletier's avatar
Vincent Pelletier committed
132 133 134
    self.exc_type = None
    self.exc_value = None
    self.traceback = None
135
    self.processing = None
136
    self.user_name = str(_getAuthenticatedUser(self))
137 138 139 140 141 142 143
    # Store REQUEST Info
    request = getattr(obj, 'REQUEST', None)
    if request is not None:
      self.request_info = dict(
        SERVER_URL=request.other['SERVER_URL'],
        VirtualRootPhysicalPath=request.other.get('VirtualRootPhysicalPath'),
        _script=list(request._script))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
144

145
  def getObject(self, activity_tool):
146
    """return the object referenced in this message."""
147
    return activity_tool.unrestrictedTraverse(self.object_path)
148

149
  def getObjectList(self, activity_tool):
150
    """return the list of object that can be expanded from this message."""
151
    object_list = []
152
    try:
153
      object_list.append(self.getObject(activity_tool))
154
    except KeyError:
155 156 157 158 159 160
      pass
    else:
      if self.hasExpandMethod():
        expand_method_id = self.activity_kw['expand_method_id']
        # FIXME: how to pass parameters?
        object_list = getattr(object_list[0], expand_method_id)()
161
    return object_list
162

163
  def hasExpandMethod(self):
164 165 166 167 168
    """return true if the message has an expand method.
    An expand method is used to expand the list of objects and to turn a
    big recursive transaction affecting many objects into multiple
    transactions affecting only one object at a time (this can prevent
    duplicated method calls)."""
169
    return self.activity_kw.has_key('expand_method_id')
170

171
  def changeUser(self, user_name, activity_tool):
172
    """restore the security context for the calling user."""
173 174
    uf = activity_tool.getPortalObject().acl_users
    user = uf.getUserById(user_name)
175
    # if the user is not found, try to get it from a parent acl_users
176 177 178 179
    # XXX this is still far from perfect, because we need to store all
    # informations about the user (like original user folder, roles) to
    # replay the activity with exactly the same security context as if
    # it had been executed without activity.
180 181 182
    if user is None:
      uf = activity_tool.getPortalObject().aq_parent.acl_users
      user = uf.getUserById(user_name)
183 184 185
    if user is not None:
      user = user.__of__(uf)
      newSecurityManager(None, user)
186
    else :
187 188
      LOG("CMFActivity", WARNING,
          "Unable to find user %s in the portal" % user_name)
189
      noSecurityManager()
190 191 192 193 194
    return user

  def activateResult(self, activity_tool, result, object):
    if self.active_process is not None:
      active_process = activity_tool.unrestrictedTraverse(self.active_process)
195
      if isinstance(result, ActiveResult):
196 197
        result.edit(object_path=object)
        result.edit(method_id=self.method_id)
198 199
        # XXX Allow other method_id in future
        active_process.activateResult(result)
200
      else:
201
        active_process.activateResult(
202
                    ActiveResult(object_path=object,
203 204
                                 method_id=self.method_id,
                                 result=result)) # XXX Allow other method_id in future
205

Jean-Paul Smets's avatar
Jean-Paul Smets committed
206
  def __call__(self, activity_tool):
207
    try:
208
      obj = self.getObject(activity_tool)
209
      old_security_manager = getSecurityManager()
210
      # Change user if required (TO BE DONE)
211
      # We will change the user only in order to execute this method
212
      user = self.changeUser(self.user_name, activity_tool)
213 214 215
      try:
        result = getattr(obj, self.method_id)(*self.args, **self.kw)
      finally:
216 217
        setSecurityManager(old_security_manager)

218
      self.activateResult(activity_tool, result, obj)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
219
      self.is_executed = 1
220
    except:
Jean-Paul Smets's avatar
Jean-Paul Smets committed
221
      self.is_executed = 0
222 223 224
      exc_info = sys.exc_info()
      self.exc_type = exc_info[0]
      self.exc_value = str(exc_info[1])
Vincent Pelletier's avatar
Vincent Pelletier committed
225
      self.traceback = ''.join(ExceptionFormatter.format_exception(
226
                               *exc_info))
227
      LOG('ActivityTool', WARNING,
228
          'Could not call method %s on object %s' % (
229
          self.method_id, self.object_path), error=exc_info)
230
      # push the error in ZODB error_log
231
      if getattr(activity_tool, 'error_log', None) is not None:
232
        activity_tool.error_log.raising(exc_info)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
233

234 235 236 237 238 239 240
  def validate(self, activity, activity_tool, check_order_validation=1):
    return activity.validate(activity_tool, self,
                             check_order_validation=check_order_validation,
                             **self.activity_kw)

  def getDependentMessageList(self, activity, activity_tool):
    return activity.getDependentMessageList(activity_tool, self, **self.activity_kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
241

242
  def notifyUser(self, activity_tool, message="Failed Processing Activity"):
243 244 245 246 247 248
    """Notify the user that the activity failed."""
    portal = activity_tool.getPortalObject()
    user_email = None
    user = portal.portal_membership.getMemberById(self.user_name)
    if user is not None:
      user_email = user.getProperty('email')
Jean-Paul Smets's avatar
Jean-Paul Smets committed
249
    if user_email in ('', None):
250 251
      user_email = portal.getProperty('email_to_address',
                       portal.getProperty('email_from_address'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
252
    mail_text = """From: %s
253 254 255 256 257 258 259
To: %s
Subject: %s

%s

Document: %s
Method: %s
Vincent Pelletier's avatar
Vincent Pelletier committed
260
Exception: %s %s
261

262
%s
263
""" % (activity_tool.email_from_address, user_email, message,
264
       message, '/'.join(self.object_path), self.method_id,
Vincent Pelletier's avatar
Vincent Pelletier committed
265
       self.exc_type, self.exc_value, self.traceback)
266 267
    try:
      activity_tool.MailHost.send( mail_text )
Vincent Pelletier's avatar
Vincent Pelletier committed
268 269
    except (socket.error, MailHostError), message:
      LOG('ActivityTool.notifyUser', WARNING, 'Mail containing failure information failed to be sent: %s. Exception was: %s %s\n%s' % (message, self.exc_type, self.exc_value, self.traceback))
270

271 272 273 274 275 276 277 278 279 280 281 282 283 284 285
  def reactivate(self, activity_tool):
    # Reactivate the original object.
    obj= self.getObject(activity_tool)
    # Change user if required (TO BE DONE)
    # We will change the user only in order to execute this method
    current_user = str(_getAuthenticatedUser(self))
    user = self.changeUser(self.user_name, activity_tool)
    try:
      active_obj = obj.activate(**self.activity_kw)
      getattr(active_obj, self.method_id)(*self.args, **self.kw)
    finally:
      # Use again the previous user
      if user is not None:
        self.changeUser(current_user, activity_tool)

Jean-Paul Smets's avatar
Jean-Paul Smets committed
286 287
class Method:

288
  def __init__(self, passive_self, activity, active_process, kw, method_id):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
289 290
    self.__passive_self = passive_self
    self.__activity = activity
291
    self.__active_process = active_process
Jean-Paul Smets's avatar
Jean-Paul Smets committed
292 293 294 295
    self.__kw = kw
    self.__method_id = method_id

  def __call__(self, *args, **kw):
296
    m = Message(self.__passive_self, self.__active_process, self.__kw, self.__method_id, args, kw)
297
    if logging:
298
      LOG('Activity Tracking', INFO, 'queuing message: activity=%s, object_path=%s, method_id=%s, args=%s, kw=%s, activity_kw=%s, user_name=%s' % (self.__activity, '/'.join(m.object_path), m.method_id, m.args, m.kw, m.activity_kw, m.user_name))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
299 300
    activity_dict[self.__activity].queueMessage(self.__passive_self.portal_activities, m)

301 302
allow_class(Method)

Jean-Paul Smets's avatar
Jean-Paul Smets committed
303 304
class ActiveWrapper:

305
  def __init__(self, passive_self, activity, active_process, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
306 307
    self.__dict__['__passive_self'] = passive_self
    self.__dict__['__activity'] = activity
308
    self.__dict__['__active_process'] = active_process
Jean-Paul Smets's avatar
Jean-Paul Smets committed
309 310 311 312
    self.__dict__['__kw'] = kw

  def __getattr__(self, id):
    return Method(self.__dict__['__passive_self'], self.__dict__['__activity'],
313
                  self.__dict__['__active_process'],
Jean-Paul Smets's avatar
Jean-Paul Smets committed
314 315
                  self.__dict__['__kw'], id)

316
class ActivityTool (Folder, UniqueObject):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
317
    """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
318 319 320 321 322 323 324 325 326 327 328 329
    ActivityTool is the central point for activity management.

    Improvement to consider to reduce locks:

      Idea 1: create an SQL tool which accumulate queries and executes them at the end of a transaction,
              thus allowing all SQL transaction to happen in a very short time
              (this would also be a great way of using MyISAM tables)

      Idea 2: do the same at the level of ActivityTool

      Idea 3: do the same at the level of each activity (ie. queueMessage
              accumulates and fires messages at the end of the transactino)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
330 331 332
    """
    id = 'portal_activities'
    meta_type = 'CMF Activity Tool'
333
    portal_type = 'Activity Tool'
334
    allowed_types = ( 'CMF Active Process', )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
335 336
    security = ClassSecurityInfo()

337 338
    manage_options = tuple(
                     [ { 'label' : 'Overview', 'action' : 'manage_overview' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
339
                     , { 'label' : 'Activities', 'action' : 'manageActivities' }
340
                     , { 'label' : 'LoadBalancing', 'action' : 'manageLoadBalancing'}
341
                     , { 'label' : 'Advanced', 'action' : 'manageActivitiesAdvanced' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
342
                     ,
343
                     ] + list(Folder.manage_options))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
344 345 346 347

    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageActivities' )
    manageActivities = DTMLFile( 'dtml/manageActivities', globals() )

348 349 350
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageActivitiesAdvanced' )
    manageActivitiesAdvanced = DTMLFile( 'dtml/manageActivitiesAdvanced', globals() )

351 352
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manage_overview' )
    manage_overview = DTMLFile( 'dtml/explainActivityTool', globals() )
353 354 355 356 357 358
    
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageLoadBalancing' )
    manageLoadBalancing = DTMLFile( 'dtml/manageLoadBalancing', globals() )
    
    distributingNode = ''
    _nodes = ()
359 360 361

    def __init__(self):
        return Folder.__init__(self, ActivityTool.id)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
362

363 364 365 366 367 368 369 370 371 372
    # Filter content (ZMI))
    def filtered_meta_types(self, user=None):
        # Filters the list of available meta types.
        all = ActivityTool.inheritedAttribute('filtered_meta_types')(self)
        meta_types = []
        for meta_type in self.all_meta_types():
            if meta_type['name'] in self.allowed_types:
                meta_types.append(meta_type)
        return meta_types

Jean-Paul Smets's avatar
Jean-Paul Smets committed
373 374
    def initialize(self):
      global is_initialized
Sebastien Robin's avatar
Sebastien Robin committed
375
      from Activity import RAMQueue, RAMDict, SQLQueue, SQLDict
Jean-Paul Smets's avatar
Jean-Paul Smets committed
376
      # Initialize each queue
377
      for activity in activity_dict.itervalues():
Jean-Paul Smets's avatar
Jean-Paul Smets committed
378
        activity.initialize(self)
Vincent Pelletier's avatar
Vincent Pelletier committed
379
      is_initialized = True
380 381 382
      
    security.declareProtected(Permissions.manage_properties, 'isSubscribed')
    def isSubscribed(self):
Aurel's avatar
Aurel committed
383
        """
384 385 386 387 388 389 390 391 392 393 394 395
        return True, if we are subscribed to TimerService.
        Otherwise return False.
        """
        service = getTimerService(self)
        if not service:
            LOG('ActivityTool', INFO, 'TimerService not available')
            return False
        
        path = '/'.join(self.getPhysicalPath())
        if path in service.lisSubscriptions():
            return True
        return False
Jean-Paul Smets's avatar
Jean-Paul Smets committed
396

397
    security.declareProtected(Permissions.manage_properties, 'subscribe')
398
    def subscribe(self, REQUEST=None, RESPONSE=None):
399 400
        """ subscribe to the global Timer Service """
        service = getTimerService(self)
401
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
402
        if not service:
403
            LOG('ActivityTool', INFO, 'TimerService not available')
404 405 406 407
            url += urllib.quote('TimerService not available')
        else:
            service.subscribe(self)
            url += urllib.quote("Subscribed to Timer Service")
408 409
        if RESPONSE is not None:
            RESPONSE.redirect(url)
410 411

    security.declareProtected(Permissions.manage_properties, 'unsubscribe')
412
    def unsubscribe(self, REQUEST=None, RESPONSE=None):
413 414
        """ unsubscribe from the global Timer Service """
        service = getTimerService(self)
415
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
416
        if not service:
417
            LOG('ActivityTool', INFO, 'TimerService not available')
418 419 420 421
            url += urllib.quote('TimerService not available')
        else:
            service.unsubscribe(self)
            url += urllib.quote("Unsubscribed from Timer Service")
422 423
        if RESPONSE is not None:
            RESPONSE.redirect(url)
424 425 426

    def manage_beforeDelete(self, item, container):
        self.unsubscribe()
427 428
        Folder.inheritedAttribute('manage_beforeDelete')(self, item, container)
    
429 430
    def manage_afterAdd(self, item, container):
        self.subscribe()
431 432
        Folder.inheritedAttribute('manage_afterAdd')(self, item, container)
       
433 434
    def getCurrentNode(self):
        """ Return current node in form ip:port """
435 436 437 438 439 440 441 442 443 444 445 446 447
        global currentNode
        if currentNode is None:
          port = ''
          from asyncore import socket_map
          for k, v in socket_map.items():
              if hasattr(v, 'port'):
                  # see Zope/lib/python/App/ApplicationManager.py: def getServers(self)
                  type = str(getattr(v, '__class__', 'unknown'))
                  if type == 'ZServer.HTTPServer.zhttp_server':
                      port = v.port
                      break
          ip = socket.gethostbyname(socket.gethostname())
          currentNode = '%s:%s' %(ip, port)
448 449 450 451 452 453 454
        return currentNode
        
    security.declarePublic('getDistributingNode')
    def getDistributingNode(self):
        """ Return the distributingNode """
        return self.distributingNode

455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491
    def getNodeList(self, role=None):
      node_dict = self.getNodeDict()
      if role is None:
        result = [x for x in node_dict.keys()]
      else:
        result = [node_id for node_id, node_role in node_dict.items() if node_role == role]
      result.sort()
      return result

    def getNodeDict(self):
      nodes = self._nodes
      if isinstance(nodes, tuple):
        new_nodes = OIBTree()
        new_nodes.update([(x, ROLE_PROCESSING) for x in self._nodes])
        self._nodes = nodes = new_nodes
      return nodes

    def registerNode(self, node):
      node_dict = self.getNodeDict()
      if not node_dict.has_key(node):
        if len(node_dict) == 0: # If we are registering the first node, make
                                # it both the distributing node and a processing
                                # node.
          role = ROLE_PROCESSING
          self.distributingNode = node
        else:
          role = ROLE_IDLE
        self.updateNode(node, role)

    def updateNode(self, node, role):
      node_dict = self.getNodeDict()
      node_dict[node] = role

    security.declareProtected(CMFCorePermissions.ManagePortal, 'getProcessingNodeList')
    def getProcessingNodeList(self):
      return self.getNodeList(role=ROLE_PROCESSING)

492
    security.declareProtected(CMFCorePermissions.ManagePortal, 'getIdleNodeList')
493 494
    def getIdleNodeList(self):
      return self.getNodeList(role=ROLE_IDLE)
495

496 497 498 499
    def _isValidNodeName(self, node_name) :
      """Check we have been provided a good node name"""
      return isinstance(node_name, str) and NODE_RE.match(node_name)
      
500 501
    security.declarePublic('manage_setDistributingNode')
    def manage_setDistributingNode(self, distributingNode, REQUEST=None):
502
        """ set the distributing node """   
503
        if not distributingNode or self._isValidNodeName(distributingNode):
504 505 506 507 508 509 510 511 512 513 514 515 516
          self.distributingNode = distributingNode
          if REQUEST is not None:
              REQUEST.RESPONSE.redirect(
                  REQUEST.URL1 +
                  '/manageLoadBalancing?manage_tabs_message=' +
                  urllib.quote("Distributing Node successfully changed."))
        else :
          if REQUEST is not None:
              REQUEST.RESPONSE.redirect(
                  REQUEST.URL1 +
                  '/manageLoadBalancing?manage_tabs_message=' +
                  urllib.quote("Malformed Distributing Node."))

517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574
    security.declareProtected(CMFCorePermissions.ManagePortal, 'manage_delNode')
    def manage_delNode(self, unused_node_list=None, REQUEST=None):
      """ delete selected unused nodes """
      processing_node = self.getDistributingNode()
      updated_processing_node = False
      if unused_node_list is not None:
        node_dict = self.getNodeDict()
        for node in unused_node_list:
          if node in node_dict:
            del node_dict[node]
          if node == processing_node:
            self.processing_node = ''
            updated_processing_node = True
      if REQUEST is not None:
        if unused_node_list is None:
          message = "No unused node selected, nothing deleted."
        else:
          message = "Deleted nodes %r." % (unused_node_list, )
        if updated_processing_node:
          message += "Disabled distributing node because it was deleted."
        REQUEST.RESPONSE.redirect(
          REQUEST.URL1 +
          '/manageLoadBalancing?manage_tabs_message=' +
          urllib.quote(message))

    security.declareProtected(CMFCorePermissions.ManagePortal, 'manage_addToProcessingList')
    def manage_addToProcessingList(self, unused_node_list=None, REQUEST=None):
      """ Change one or more idle nodes into processing nodes """
      if unused_node_list is not None:
        node_dict = self.getNodeDict()
        for node in unused_node_list:
          self.updateNode(node, ROLE_PROCESSING)
      if REQUEST is not None:
        if unused_node_list is None:
          message = "No unused node selected, nothing done."
        else:
          message = "Nodes now procesing: %r." % (unused_node_list, )
        REQUEST.RESPONSE.redirect(
          REQUEST.URL1 +
          '/manageLoadBalancing?manage_tabs_message=' +
          urllib.quote(message))

    security.declareProtected(CMFCorePermissions.ManagePortal, 'manage_removeFromProcessingList')
    def manage_removeFromProcessingList(self, processing_node_list=None, REQUEST=None):
      """ Change one or more procesing nodes into idle nodes """
      if processing_node_list is not None:
        node_dict = self.getNodeDict()
        for node in processing_node_list:
          self.updateNode(node, ROLE_IDLE)
      if REQUEST is not None:
        if processing_node_list is None:
          message = "No used node selected, nothing done."
        else:
          message = "Nodes now unused %r." % (processing_node_list, )
        REQUEST.RESPONSE.redirect(
          REQUEST.URL1 +
          '/manageLoadBalancing?manage_tabs_message=' +
          urllib.quote(message))
575

576
    def process_timer(self, tick, interval, prev="", next=""):
577
        """
578 579 580 581 582
        Call distribute() if we are the Distributing Node and call tic()
        with our node number.
        This method is called by TimerService in the interval given
        in zope.conf. The Default is every 5 seconds.
        """
583 584 585 586
        # Prevent TimerService from starting multiple threads in parallel
        acquired = timerservice_lock.acquire(0)
        if not acquired:
          return
587

588
        try:
589
          old_sm = getSecurityManager()
590
          try:
591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622
            try:
              # get owner of portal_catalog, so normally we should be able to
              # have the permission to invoke all activities
              user = self.portal_catalog.getWrappedOwner()
              newSecurityManager(self.REQUEST, user)

              currentNode = self.getCurrentNode()
              self.registerNode(currentNode)
              processing_node_list = self.getNodeList(role=ROLE_PROCESSING)

              # only distribute when we are the distributingNode or if it's empty
              if (self.getDistributingNode() == currentNode):
                self.distribute(len(processing_node_list))

              # SkinsTool uses a REQUEST cache to store skin objects, as
              # with TimerService we have the same REQUEST over multiple
              # portals, we clear this cache to make sure the cache doesn't
              # contains skins from another portal.
              stool = getToolByName(self, 'portal_skins', None)
              if stool is not None:
                stool.changeSkin(None)

              # call tic for the current processing_node
              # the processing_node numbers are the indices of the elements in the node tuple +1
              # because processing_node starts form 1
              if currentNode in processing_node_list:
                self.tic(processing_node_list.index(currentNode)+1)
            except:
              # Catch ALL exception to avoid killing timerserver.
              LOG('ActivityTool', ERROR, 'process_timer received an exception', error=sys.exc_info())
          finally:
            setSecurityManager(old_sm)
Jérome Perrin's avatar
Jérome Perrin committed
623
        finally:
624
          timerservice_lock.release()
625

Jean-Paul Smets's avatar
Jean-Paul Smets committed
626 627 628 629 630 631
    security.declarePublic('distribute')
    def distribute(self, node_count=1):
      """
        Distribute load
      """
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
632 633
      if not is_initialized:
        self.initialize()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
634 635

      # Call distribute on each queue
636
      for activity in activity_dict.itervalues():
637
        activity.distribute(aq_inner(self), node_count)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
638

Jean-Paul Smets's avatar
Jean-Paul Smets committed
639
    security.declarePublic('tic')
Jean-Paul Smets's avatar
Jean-Paul Smets committed
640
    def tic(self, processing_node=1, force=0):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
641 642
      """
        Starts again an activity
Jean-Paul Smets's avatar
Jean-Paul Smets committed
643
        processing_node starts from 1 (there is not node 0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
644
      """
Vincent Pelletier's avatar
Vincent Pelletier committed
645
      global active_threads, first_run
Jean-Paul Smets's avatar
Jean-Paul Smets committed
646 647

      # return if the number of threads is too high
648
      # else, increase the number of active_threads and continue
649 650
      tic_lock.acquire()
      too_many_threads = (active_threads >= max_active_threads)
651
      if not too_many_threads or force:
652
        active_threads += 1
653 654 655
      else:
        tic_lock.release()
        raise RuntimeError, 'Too many threads'
656
      tic_lock.release()
657

Jean-Paul Smets's avatar
Jean-Paul Smets committed
658
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
659 660
      if not is_initialized:
        self.initialize()
661

662
      inner_self = aq_inner(self)
663

664 665 666
      # If this is the first tic after zope is started, reset the processing
      # flag for activities of this node
      if first_run:
667 668 669 670
        inner_self.SQLDict_clearProcessingFlag(
                                processing_node=processing_node)
        inner_self.SQLQueue_clearProcessingFlag(
                                processing_node=processing_node)
671
        first_run = False
672

673
      try:
674 675 676 677 678 679 680
        #Sort activity list by priority
        activity_list = activity_dict.values()
        # Sort method must be local to access "self"
        def cmpActivities(activity_1, activity_2):
          return cmp(activity_1.getPriority(self), activity_2.getPriority(self))
        activity_list.sort(cmpActivities)
        
681
        # Wakeup each queue
682
        for activity in activity_list:
683
          activity.wakeup(inner_self, processing_node)
684

685 686 687 688
        # Process messages on each queue in round robin
        has_awake_activity = 1
        while has_awake_activity:
          has_awake_activity = 0
689
          for activity in activity_list:
690 691
            activity.tic(inner_self, processing_node) # Transaction processing is the responsability of the activity
            has_awake_activity = has_awake_activity or activity.isAwake(inner_self, processing_node)
692 693 694 695 696
      finally:
        # decrease the number of active_threads
        tic_lock.acquire()
        active_threads -= 1
        tic_lock.release()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
697

698
    def hasActivity(self, *args, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
699
      # Check in each queue if the object has deferred tasks
700 701
      # if not argument is provided, then check on self
      if len(args) > 0:
702
        obj = args[0]
703
      else:
704
        obj = self
705
      for activity in activity_dict.itervalues():
706
        if activity.hasActivity(aq_inner(self), obj, **kw):
707 708
          return True
      return False
Jean-Paul Smets's avatar
Jean-Paul Smets committed
709

710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725
    def getActivityBuffer(self, create_if_not_found=True):
      """
        Get activtity buffer for this thread for this activity tool.
        If no activity buffer is found at lowest level and create_if_not_found
        is True, create one.
        Intermediate level is unconditionaly created if non existant because
        chances are it will be used in the instance life.
        Lock is held when checking for intermediate level existance
        because:
         - intermediate level dict must not be created in 2 threads at the
           same time, since one creation would destroy the existing one.
        It's released after that step because:
         - lower level access is at thread scope, thus by definition there
           can be only one access at a time to a key
         - GIL protects us when accessing python instances
      """
726 727
      # Safeguard: make sure we are wrapped in  acquisition context before
      # using our path as an activity tool instance-wide identifier.
728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746
      assert getattr(self, 'aq_self', None) is not None
      my_instance_key = self.getPhysicalPath()
      my_thread_key = get_ident()
      global_activity_buffer_lock.acquire()
      try:
        if my_instance_key not in global_activity_buffer:
          global_activity_buffer[my_instance_key] = {}
      finally:
        global_activity_buffer_lock.release()
      thread_activity_buffer = global_activity_buffer[my_instance_key]
      if my_thread_key not in thread_activity_buffer:
        if create_if_not_found:
          buffer = ActivityBuffer(activity_tool=self)
        else:
          buffer = None
        thread_activity_buffer[my_thread_key] = buffer
      activity_buffer = thread_activity_buffer[my_thread_key]
      return activity_buffer

747 748
    security.declarePrivate('activateObject')
    def activateObject(self, object, activity, active_process, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
749 750
      if not is_initialized:
        self.initialize()
751
      self.getActivityBuffer()
752
      return ActiveWrapper(object, activity, active_process, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
753

754
    def deferredQueueMessage(self, activity, message):
755 756
      activity_buffer = self.getActivityBuffer()
      activity_buffer.deferredQueueMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
757

758
    def deferredDeleteMessage(self, activity, message):
759 760
      activity_buffer = self.getActivityBuffer()
      activity_buffer.deferredDeleteMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
761

Jean-Paul Smets's avatar
Jean-Paul Smets committed
762
    def getRegisteredMessageList(self, activity):
763
      activity_buffer = self.getActivityBuffer(create_if_not_found=False)
764
      if activity_buffer is not None:
765 766
        #activity_buffer._register() # This is required if flush flush is called outside activate
        return activity.getRegisteredMessageList(activity_buffer,
767
                                                 aq_inner(self))
768 769
      else:
        return []
Yoshinori Okuji's avatar
Yoshinori Okuji committed
770

Jean-Paul Smets's avatar
Jean-Paul Smets committed
771
    def unregisterMessage(self, activity, message):
772 773 774
      activity_buffer = self.getActivityBuffer()
      #activity_buffer._register()
      return activity.unregisterMessage(activity_buffer, aq_inner(self), message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
775

776
    def flush(self, obj, invoke=0, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
777 778
      if not is_initialized:
        self.initialize()
779
      self.getActivityBuffer()
780 781
      if isinstance(obj, tuple):
        object_path = obj
782
      else:
783
        object_path = obj.getPhysicalPath()
784
      for activity in activity_dict.itervalues():
785
        activity.flush(aq_inner(self), object_path, invoke=invoke, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
786

787
    def start(self, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
788 789
      if not is_initialized:
        self.initialize()
790
      for activity in activity_dict.itervalues():
791
        activity.start(aq_inner(self), **kw)
792 793

    def stop(self, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
794 795
      if not is_initialized:
        self.initialize()
796
      for activity in activity_dict.itervalues():
797
        activity.stop(aq_inner(self), **kw)
798

Jean-Paul Smets's avatar
Jean-Paul Smets committed
799
    def invoke(self, message):
800
      if logging:
801
        LOG('Activity Tracking', INFO, 'invoking message: object_path=%s, method_id=%s, args=%s, kw=%s, activity_kw=%s, user_name=%s' % ('/'.join(message.object_path), message.method_id, message.args, message.kw, message.activity_kw, message.user_name))
802 803
      if getattr(self, 'aq_chain', None) is not None:
        # Grab existing acquisition chain and extrach base objects.
804
        base_chain = [aq_base(x) for x in self.aq_chain]
805 806 807 808 809 810 811
        # Grab existig request (last chain item) and create a copy.
        request_container = base_chain.pop()
        request = request_container.REQUEST
        # XXX: REQUEST.clone() requires PARENTS to be set, and it's not when
        # runing unit tests. Recreate it if it does not exist.
        parents = getattr(request, 'PARENTS', None)
        if parents is None:
812
          warn('CMFActivity.ActivityTool.invoke: PARENTS is not defined in REQUEST. It should only happen in unit tests.')
813
          request['PARENTS'] = self.aq_chain[:]
814 815 816 817 818 819 820 821 822 823 824 825
        
        # restore request information
        new_request = request.clone()
        request_info = getattr(message, 'request_info', None)
        if request_info is not None:
          new_request.other['SERVER_URL'] = request_info['SERVER_URL']
          virtual_root_path = request_info.get('VirtualRootPhysicalPath')
          if virtual_root_path:
            new_request.other['VirtualRootPhysicalPath'] = virtual_root_path
          new_request._script = request_info['_script']

        new_request_container = request_container.__class__(REQUEST=new_request)
826 827 828 829 830 831 832 833 834
        # Recreate acquisition chain.
        my_self = new_request_container
        base_chain.reverse()
        for item in base_chain:
          my_self = item.__of__(my_self)
      else:
        my_self = self
        LOG('CMFActivity.ActivityTool.invoke', INFO, 'Strange: invoke is called outside of acquisition context.')
      message(my_self)
835
      if logging:
836
        LOG('Activity Tracking', INFO, 'invoked message')
837 838 839
      if my_self is not self: # We rewrapped self
        for held in my_self.REQUEST._held:
          self.REQUEST._hold(held)
840

841
    def invokeGroup(self, method_id, message_list):
842
      if logging:
843
        LOG('Activity Tracking', INFO, 'invoking group messages: method_id=%s, paths=%s' % (method_id, ['/'.join(m.object_path) for m in message_list]))
844 845 846 847 848 849 850 851
      # Invoke a group method.
      object_list = []
      expanded_object_list = []
      new_message_list = []
      path_dict = {}
      # Filter the list of messages. If an object is not available, ignore such a message.
      # In addition, expand an object if necessary, and make sure that no duplication happens.
      for m in message_list:
852 853
        # alternate method is used to segregate objects which cannot be grouped.
        alternate_method_id = m.activity_kw.get('alternate_method_id')
854 855
        try:
          obj = m.getObject(self)
856
          i = len(new_message_list) # This is an index of this message in new_message_list.
857
          if m.hasExpandMethod():
858 859
            for subobj in m.getObjectList(self):
              path = subobj.getPath()
860
              if path not in path_dict:
861
                path_dict[path] = i
862 863 864 865 866 867
                if alternate_method_id is not None \
                   and hasattr(aq_base(subobj), alternate_method_id):
                  # if this object is alternated, generate a new single active object.
                  activity_kw = m.activity_kw.copy()
                  if 'group_method_id' in activity_kw:
                    del activity_kw['group_method_id']
868 869
                  if 'group_id' in activity_kw:
                    del activity_kw['group_id']                    
870 871 872 873
                  active_obj = subobj.activate(**activity_kw)
                  getattr(active_obj, alternate_method_id)(*m.args, **m.kw)
                else:
                  expanded_object_list.append(subobj)
874 875 876
          else:
            path = obj.getPath()
            if path not in path_dict:
877
              path_dict[path] = i
878 879 880 881 882 883
              if alternate_method_id is not None \
                  and hasattr(aq_base(obj), alternate_method_id):
                # if this object is alternated, generate a new single active object.
                activity_kw = m.activity_kw.copy()
                if 'group_method_id' in activity_kw:
                  del activity_kw['group_method_id']
884 885
                if 'group_id' in activity_kw:
                  del activity_kw['group_id']
886 887 888 889
                active_obj = obj.activate(**activity_kw)
                getattr(active_obj, alternate_method_id)(*m.args, **m.kw)
              else:
                expanded_object_list.append(obj)
890
          object_list.append(obj)
891 892 893
          new_message_list.append(m)
        except:
          m.is_executed = 0
894 895 896 897 898
          exc_info = sys.exc_info()
          m.exc_type = exc_info[0]
          m.exc_value = str(exc_info[1])
          m.traceback = ''.join(ExceptionFormatter.format_exception(
                                *exc_info))
899
          LOG('WARNING ActivityTool', 0,
900
              'Could not call method %s on object %s' %
901
              (m.method_id, m.object_path), error=exc_info)
902

903 904
      try:
        if len(expanded_object_list) > 0:
905 906
          method = self.unrestrictedTraverse(method_id)
          # FIXME: how to apply security here?
907 908
          # NOTE: expanded_object_list must be set to failed objects by the callee.
          #       If it fully succeeds, expanded_object_list must be empty when returning.
909
          result = method(expanded_object_list, **m.kw)
910
        else:
911 912 913
          result = None
      except:
        # In this case, the group method completely failed.
914 915 916 917 918
        exc_info = sys.exc_info()
        exc_type = exc_info[0]
        exc_value = str(exc_info[1])
        traceback = ''.join(ExceptionFormatter.format_exception(
                            *exc_info))
919 920
        for m in new_message_list:
          m.is_executed = 0
921 922 923
          m.exc_type = exc_type
          m.exc_value = exc_value
          m.traceback = traceback
924
        LOG('WARNING ActivityTool', 0,
925
            'Could not call method %s on objects %s' %
926
            (method_id, expanded_object_list), error=exc_info)
927 928 929 930 931 932 933 934 935 936 937 938 939 940 941
      else:
        # Obtain all indices of failed messages. Note that this can be a partial failure.
        failed_message_dict = {}
        for obj in expanded_object_list:
          path = obj.getPath()
          i = path_dict[path]
          failed_message_dict[i] = None

        # Only for succeeded messages, an activity process is invoked (if any).
        for i in xrange(len(object_list)):
          object = object_list[i]
          m = new_message_list[i]
          if i in failed_message_dict:
            m.is_executed = 0
            LOG('ActivityTool', WARNING,
942 943
                'the method %s partially failed on object %s' %
                (m.method_id, m.object_path,))
944 945 946 947 948
          else:
            try:
              m.activateResult(self, result, object)
              m.is_executed = 1
            except:
949
              m.is_executed = 0
Vincent Pelletier's avatar
Vincent Pelletier committed
950
              m.exc_type = sys.exc_info()[0]
951
              LOG('ActivityTool', WARNING,
952
                  'Could not call method %s on object %s' % (
Vincent Pelletier's avatar
Vincent Pelletier committed
953
                  m.method_id, m.object_path), error=sys.exc_info())
954
      if logging:
955
        LOG('Activity Tracking', INFO, 'invoked group messages')
956

957 958
    def newMessage(self, activity, path, active_process,
                   activity_kw, method_id, *args, **kw):
959
      # Some Security Cheking should be made here XXX
Vincent Pelletier's avatar
Vincent Pelletier committed
960 961
      if not is_initialized:
        self.initialize()
962
      self.getActivityBuffer()
963
      activity_dict[activity].queueMessage(aq_inner(self),
964
        Message(path, active_process, activity_kw, method_id, args, kw))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
965

966
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageInvoke' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
967 968 969 970 971 972
    def manageInvoke(self, object_path, method_id, REQUEST=None):
      """
        Invokes all methods for object "object_path"
      """
      if type(object_path) is type(''):
        object_path = tuple(object_path.split('/'))
973
      self.flush(object_path,method_id=method_id,invoke=1)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
974
      if REQUEST is not None:
975 976
        return REQUEST.RESPONSE.redirect('%s/%s' %
                (self.absolute_url(), 'manageActivities'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
977

978
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageCancel' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
979 980 981 982 983 984
    def manageCancel(self, object_path, method_id, REQUEST=None):
      """
        Cancel all methods for object "object_path"
      """
      if type(object_path) is type(''):
        object_path = tuple(object_path.split('/'))
985
      self.flush(object_path,method_id=method_id,invoke=0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
986
      if REQUEST is not None:
987 988
        return REQUEST.RESPONSE.redirect('%s/%s' %
                (self.absolute_url(), 'manageActivities'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
989

990 991
    security.declareProtected( CMFCorePermissions.ManagePortal,
                               'manageClearActivities' )
992
    def manageClearActivities(self, keep=1, REQUEST=None):
993 994 995 996 997
      """
        Clear all activities and recreate tables.
      """
      folder = getToolByName(self, 'portal_skins').activity

998 999
      # Obtain all pending messages.
      message_list = []
1000
      if keep:
1001
        for activity in activity_dict.itervalues():
1002 1003 1004 1005 1006 1007
          if hasattr(activity, 'dumpMessageList'):
            try:
              message_list.extend(activity.dumpMessageList(self))
            except ConflictError:
              raise
            except:
1008 1009 1010
              LOG('ActivityTool', WARNING,
                  'could not dump messages from %s' %
                  (activity,), error=sys.exc_info())
1011 1012

      if getattr(folder, 'SQLDict_createMessageTable', None) is not None:
1013 1014 1015 1016 1017
        try:
          folder.SQLDict_dropMessageTable()
        except ConflictError:
          raise
        except:
1018
          LOG('CMFActivity', WARNING,
1019
              'could not drop the message table',
1020 1021 1022
              error=sys.exc_info())
        folder.SQLDict_createMessageTable()

1023
      if getattr(folder, 'SQLQueue_createMessageTable', None) is not None:
1024 1025 1026 1027 1028
        try:
          folder.SQLQueue_dropMessageTable()
        except ConflictError:
          raise
        except:
1029
          LOG('CMFActivity', WARNING,
1030
              'could not drop the message queue table',
1031 1032 1033
              error=sys.exc_info())
        folder.SQLQueue_createMessageTable()

1034 1035 1036
      # Reactivate the messages.
      for m in message_list:
        try:
1037
          m.reactivate(aq_inner(self))
1038 1039 1040 1041
        except ConflictError:
          raise
        except:
          LOG('ActivityTool', WARNING,
1042 1043
              'could not reactivate the message %r, %r' %
              (m.object_path, m.method_id), error=sys.exc_info())
1044

1045
      if REQUEST is not None:
1046 1047 1048 1049 1050 1051
        message = 'Activities%20Cleared'
        if keep:
          message = 'Tables%20Recreated'
        return REQUEST.RESPONSE.redirect(
            '%s/manageActivitiesAdvanced?manage_tabs_message=%s' % (
              self.absolute_url(), message))
1052

Jean-Paul Smets's avatar
Jean-Paul Smets committed
1053
    security.declarePublic('getMessageList')
1054
    def getMessageList(self,**kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1055 1056 1057
      """
        List messages waiting in queues
      """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1058
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
1059 1060
      if not is_initialized:
        self.initialize()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1061

Jean-Paul Smets's avatar
Jean-Paul Smets committed
1062
      message_list = []
1063
      for activity in activity_dict.itervalues():
Sebastien Robin's avatar
Sebastien Robin committed
1064
        try:
1065
          message_list += activity.getMessageList(aq_inner(self),**kw)
Sebastien Robin's avatar
Sebastien Robin committed
1066 1067
        except AttributeError:
          LOG('getMessageList, could not get message from Activity:',0,activity)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1068 1069
      return message_list

1070 1071 1072 1073 1074 1075
    security.declarePublic('countMessageWithTag')
    def countMessageWithTag(self, value):
      """
        Return the number of messages which match the given tag.
      """
      message_count = 0
1076
      for activity in activity_dict.itervalues():
1077
        message_count += activity.countMessageWithTag(aq_inner(self), value)
Sebastien Robin's avatar
Sebastien Robin committed
1078 1079 1080 1081 1082 1083 1084 1085 1086 1087
      return message_count

    security.declarePublic('countMessage')
    def countMessage(self, **kw):
      """
        Return the number of messages which match the given parameter.

        Parameters allowed:

        method_id : the id of the method
Jérome Perrin's avatar
Jérome Perrin committed
1088
        path : for activities on a particular object
Sebastien Robin's avatar
Sebastien Robin committed
1089 1090 1091 1092
        tag : activities with a particular tag
        message_uid : activities with a particular uid
      """
      message_count = 0
1093
      for activity in activity_dict.itervalues():
1094
        message_count += activity.countMessage(aq_inner(self), **kw)
1095 1096
      return message_count

1097
    security.declareProtected( CMFCorePermissions.ManagePortal , 'newActiveProcess' )
1098
    def newActiveProcess(self, **kw):
1099 1100 1101
      from ActiveProcess import addActiveProcess
      new_id = str(self.generateNewId())
      addActiveProcess(self, new_id)
1102 1103 1104
      active_process = self._getOb(new_id)
      active_process.edit(**kw)
      return active_process
1105 1106 1107 1108

    def reindexObject(self):
      self.immediateReindexObject()

1109
    # Active synchronisation methods
1110
    security.declarePrivate('validateOrder')
1111
    def validateOrder(self, message, validator_id, validation_value):
1112 1113 1114 1115 1116
      message_list = self.getDependentMessageList(message, validator_id, validation_value)
      return len(message_list) > 0

    security.declarePrivate('getDependentMessageList')
    def getDependentMessageList(self, message, validator_id, validation_value):
Vincent Pelletier's avatar
Vincent Pelletier committed
1117 1118
      if not is_initialized:
        self.initialize()
1119
      message_list = []
Vincent Pelletier's avatar
Vincent Pelletier committed
1120
      method_id = "_validate_%s" % validator_id
1121
      for activity in activity_dict.itervalues():
1122 1123 1124 1125 1126 1127
        method = getattr(activity, method_id, None)
        if method is not None:
          result = method(aq_inner(self), message, validation_value)
          if result:
            message_list.extend([(activity, m) for m in result])
      return message_list
1128

Yoshinori Okuji's avatar
Yoshinori Okuji committed
1129 1130
    # Required for tests (time shift)
    def timeShift(self, delay):
Vincent Pelletier's avatar
Vincent Pelletier committed
1131 1132
      if not is_initialized:
        self.initialize()
1133
      for activity in activity_dict.itervalues():
1134
        activity.timeShift(aq_inner(self), delay)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
1135

1136
InitializeClass(ActivityTool)