ActivityTool.py 46.7 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
from Products import iHotfix
53

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

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

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

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

Jean-Paul Smets's avatar
Jean-Paul Smets committed
69 70 71 72
# 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
73
is_initialized = False
74 75
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
76
first_run = True
77 78 79
currentNode = None
ROLE_IDLE = 0
ROLE_PROCESSING = 1
Jean-Paul Smets's avatar
Jean-Paul Smets committed
80 81 82 83

# Activity Registration
activity_dict = {}

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

def enableLogging():
  global logging
  logging = True

def disableLogging():
  global logging
  logging = False

94 95 96 97 98 99 100
# 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
101 102 103
def registerActivity(activity):
  # Must be rewritten to register
  # class and create instance for each activity
104
  #LOG('Init Activity', 0, str(activity.__name__))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
105 106 107
  activity_instance = activity()
  activity_dict[activity.__name__] = activity_instance

108 109 110 111
MESSAGE_NOT_EXECUTED = 0
MESSAGE_EXECUTED = 1
MESSAGE_NOT_EXECUTABLE = 2

Jean-Paul Smets's avatar
Jean-Paul Smets committed
112
class Message:
113
  """Activity Message Class.
114

115 116
  Message instances are stored in an activity queue, inside the Activity Tool.
  """
117 118
  def __init__(self, obj, active_process, activity_kw, method_id, args, kw):
    if isinstance(obj, str):
119
      self.object_path = tuple(obj.split('/'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
120
    else:
121
      self.object_path = obj.getPhysicalPath()
Yoshinori Okuji's avatar
Yoshinori Okuji committed
122
    if type(active_process) is StringType:
123 124 125 126 127
      self.active_process = active_process.split('/')
    elif active_process is None:
      self.active_process = None
    else:
      self.active_process = active_process.getPhysicalPath()
128
      self.active_process_uid = active_process.getUid()
129 130 131
    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
132 133 134 135
    self.activity_kw = activity_kw
    self.method_id = method_id
    self.args = args
    self.kw = kw
136
    self.is_executed = MESSAGE_NOT_EXECUTED
Vincent Pelletier's avatar
Vincent Pelletier committed
137 138 139
    self.exc_type = None
    self.exc_value = None
    self.traceback = None
140
    self.processing = None
141
    self.user_name = str(_getAuthenticatedUser(self))
142
    # Store REQUEST Info
143
    self.request_info = {}
144 145
    request = getattr(obj, 'REQUEST', None)
    if request is not None:
146 147 148 149 150 151 152 153 154
      if 'SERVER_URL' in request.other:
        self.request_info['SERVER_URL'] = request.other['SERVER_URL']
      if 'VirtualRootPhysicalPath' in request.other:
        self.request_info['VirtualRootPhysicalPath'] = \
          request.other['VirtualRootPhysicalPath']
      if 'HTTP_ACCEPT_LANGUAGE' in request.environ:
        self.request_info['HTTP_ACCEPT_LANGUAGE'] = \
          request.environ['HTTP_ACCEPT_LANGUAGE']
      self.request_info['_script'] = list(request._script)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
155

156
  def getObject(self, activity_tool):
157
    """return the object referenced in this message."""
158
    return activity_tool.unrestrictedTraverse(self.object_path)
159

160
  def getObjectList(self, activity_tool):
161
    """return the list of object that can be expanded from this message."""
162
    object_list = []
163
    try:
164
      object_list.append(self.getObject(activity_tool))
165
    except KeyError:
166 167 168 169 170 171
      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)()
172
    return object_list
173

174
  def hasExpandMethod(self):
175 176 177 178 179
    """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)."""
180
    return self.activity_kw.has_key('expand_method_id')
181

182
  def changeUser(self, user_name, activity_tool):
183
    """restore the security context for the calling user."""
184 185
    uf = activity_tool.getPortalObject().acl_users
    user = uf.getUserById(user_name)
186
    # if the user is not found, try to get it from a parent acl_users
187 188 189 190
    # 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.
191 192 193
    if user is None:
      uf = activity_tool.getPortalObject().aq_parent.acl_users
      user = uf.getUserById(user_name)
194 195 196
    if user is not None:
      user = user.__of__(uf)
      newSecurityManager(None, user)
197
    else :
198 199
      LOG("CMFActivity", WARNING,
          "Unable to find user %s in the portal" % user_name)
200
      noSecurityManager()
201 202 203 204 205
    return user

  def activateResult(self, activity_tool, result, object):
    if self.active_process is not None:
      active_process = activity_tool.unrestrictedTraverse(self.active_process)
206
      if isinstance(result, ActiveResult):
207 208
        result.edit(object_path=object)
        result.edit(method_id=self.method_id)
209 210
        # XXX Allow other method_id in future
        active_process.activateResult(result)
211
      else:
212
        active_process.activateResult(
213
                    ActiveResult(object_path=object,
214 215
                                 method_id=self.method_id,
                                 result=result)) # XXX Allow other method_id in future
216

Jean-Paul Smets's avatar
Jean-Paul Smets committed
217
  def __call__(self, activity_tool):
218
    try:
219
      obj = self.getObject(activity_tool)
220 221 222
    except KeyError:
      self.is_executed = MESSAGE_NOT_EXECUTABLE
    else:
223
      try:
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254
        old_security_manager = getSecurityManager()
        # Change user if required (TO BE DONE)
        # We will change the user only in order to execute this method
        user = self.changeUser(self.user_name, activity_tool)
        # XXX: There is no check to see if user is allowed to access
        # that method !
        method = getattr(obj, self.method_id, None)
        try:
          if method is None:
            self.is_executed = MESSAGE_NOT_EXECUTABLE
          else:
            result = method(*self.args, **self.kw)
        finally:
          setSecurityManager(old_security_manager)

        if method is not None:
          self.activateResult(activity_tool, result, obj)
          self.is_executed = MESSAGE_EXECUTED
      except:
        self.is_executed = MESSAGE_NOT_EXECUTED
        exc_info = sys.exc_info()
        self.exc_type = exc_info[0]
        self.exc_value = str(exc_info[1])
        self.traceback = ''.join(ExceptionFormatter.format_exception(
                                 *exc_info))
        LOG('ActivityTool', WARNING,
            'Could not call method %s on object %s' % (
            self.method_id, self.object_path), error=exc_info)
        # push the error in ZODB error_log
        if getattr(activity_tool, 'error_log', None) is not None:
          activity_tool.error_log.raising(exc_info)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
255

256 257 258 259 260 261 262
  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
263

264
  def notifyUser(self, activity_tool, message="Failed Processing Activity"):
265 266
    """Notify the user that the activity failed."""
    portal = activity_tool.getPortalObject()
267
    user_email = portal.getProperty('email_to_address',
268
                       portal.getProperty('email_from_address'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
269
    mail_text = """From: %s
270 271 272 273 274 275 276
To: %s
Subject: %s

%s

Document: %s
Method: %s
Vincent Pelletier's avatar
Vincent Pelletier committed
277
Exception: %s %s
278

279
%s
280
""" % (activity_tool.email_from_address, user_email, message,
281
       message, '/'.join(self.object_path), self.method_id,
Vincent Pelletier's avatar
Vincent Pelletier committed
282
       self.exc_type, self.exc_value, self.traceback)
283 284
    try:
      activity_tool.MailHost.send( mail_text )
Vincent Pelletier's avatar
Vincent Pelletier committed
285 286
    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))
287

288 289 290 291 292 293 294 295 296 297 298 299 300 301 302
  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
303 304
class Method:

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

  def __call__(self, *args, **kw):
313
    m = Message(self.__passive_self, self.__active_process, self.__kw, self.__method_id, args, kw)
314
    if logging:
315
      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
316 317
    activity_dict[self.__activity].queueMessage(self.__passive_self.portal_activities, m)

318 319
allow_class(Method)

Jean-Paul Smets's avatar
Jean-Paul Smets committed
320 321
class ActiveWrapper:

322
  def __init__(self, passive_self, activity, active_process, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
323 324
    self.__dict__['__passive_self'] = passive_self
    self.__dict__['__activity'] = activity
325
    self.__dict__['__active_process'] = active_process
Jean-Paul Smets's avatar
Jean-Paul Smets committed
326 327 328 329
    self.__dict__['__kw'] = kw

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

333
class ActivityTool (Folder, UniqueObject):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
334
    """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
335 336 337 338 339 340 341 342 343 344 345 346
    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
347 348 349
    """
    id = 'portal_activities'
    meta_type = 'CMF Activity Tool'
350
    portal_type = 'Activity Tool'
351
    allowed_types = ( 'CMF Active Process', )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
352 353
    security = ClassSecurityInfo()

354 355
    manage_options = tuple(
                     [ { 'label' : 'Overview', 'action' : 'manage_overview' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
356
                     , { 'label' : 'Activities', 'action' : 'manageActivities' }
357
                     , { 'label' : 'LoadBalancing', 'action' : 'manageLoadBalancing'}
358
                     , { 'label' : 'Advanced', 'action' : 'manageActivitiesAdvanced' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
359
                     ,
360
                     ] + list(Folder.manage_options))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
361 362 363 364

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

365 366 367
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageActivitiesAdvanced' )
    manageActivitiesAdvanced = DTMLFile( 'dtml/manageActivitiesAdvanced', globals() )

368 369
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manage_overview' )
    manage_overview = DTMLFile( 'dtml/explainActivityTool', globals() )
370 371 372 373 374 375
    
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageLoadBalancing' )
    manageLoadBalancing = DTMLFile( 'dtml/manageLoadBalancing', globals() )
    
    distributingNode = ''
    _nodes = ()
376 377 378

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

380 381 382 383 384 385 386 387 388 389
    # 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
390 391
    def initialize(self):
      global is_initialized
Sebastien Robin's avatar
Sebastien Robin committed
392
      from Activity import RAMQueue, RAMDict, SQLQueue, SQLDict
Jean-Paul Smets's avatar
Jean-Paul Smets committed
393
      # Initialize each queue
394
      for activity in activity_dict.itervalues():
Jean-Paul Smets's avatar
Jean-Paul Smets committed
395
        activity.initialize(self)
Vincent Pelletier's avatar
Vincent Pelletier committed
396
      is_initialized = True
397 398 399
      
    security.declareProtected(Permissions.manage_properties, 'isSubscribed')
    def isSubscribed(self):
Aurel's avatar
Aurel committed
400
        """
401 402 403 404 405 406 407 408 409 410 411 412
        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
413

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

    security.declareProtected(Permissions.manage_properties, 'unsubscribe')
429
    def unsubscribe(self, REQUEST=None, RESPONSE=None):
430 431
        """ unsubscribe from the global Timer Service """
        service = getTimerService(self)
432
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
433
        if not service:
434
            LOG('ActivityTool', INFO, 'TimerService not available')
435 436 437 438
            url += urllib.quote('TimerService not available')
        else:
            service.unsubscribe(self)
            url += urllib.quote("Unsubscribed from Timer Service")
439 440
        if RESPONSE is not None:
            RESPONSE.redirect(url)
441 442 443

    def manage_beforeDelete(self, item, container):
        self.unsubscribe()
444 445
        Folder.inheritedAttribute('manage_beforeDelete')(self, item, container)
    
446 447
    def manage_afterAdd(self, item, container):
        self.subscribe()
448 449
        Folder.inheritedAttribute('manage_afterAdd')(self, item, container)
       
450 451
    def getCurrentNode(self):
        """ Return current node in form ip:port """
452 453 454 455 456 457 458 459 460 461 462 463 464
        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)
465 466 467 468 469 470 471
        return currentNode
        
    security.declarePublic('getDistributingNode')
    def getDistributingNode(self):
        """ Return the distributingNode """
        return self.distributingNode

472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508
    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)

509
    security.declareProtected(CMFCorePermissions.ManagePortal, 'getIdleNodeList')
510 511
    def getIdleNodeList(self):
      return self.getNodeList(role=ROLE_IDLE)
512

513 514 515 516
    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)
      
517 518
    security.declarePublic('manage_setDistributingNode')
    def manage_setDistributingNode(self, distributingNode, REQUEST=None):
519
        """ set the distributing node """   
520
        if not distributingNode or self._isValidNodeName(distributingNode):
521 522 523 524 525 526 527 528 529 530 531 532 533
          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."))

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 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591
    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))
592

593
    def process_timer(self, tick, interval, prev="", next=""):
594
        """
595 596 597 598 599
        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.
        """
600 601 602 603
        # Prevent TimerService from starting multiple threads in parallel
        acquired = timerservice_lock.acquire(0)
        if not acquired:
          return
604

605
        try:
606
          old_sm = getSecurityManager()
607
          try:
608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639
            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
640
        finally:
641
          timerservice_lock.release()
642

Jean-Paul Smets's avatar
Jean-Paul Smets committed
643 644 645 646 647 648
    security.declarePublic('distribute')
    def distribute(self, node_count=1):
      """
        Distribute load
      """
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
649 650
      if not is_initialized:
        self.initialize()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
651 652

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

Jean-Paul Smets's avatar
Jean-Paul Smets committed
656
    security.declarePublic('tic')
Jean-Paul Smets's avatar
Jean-Paul Smets committed
657
    def tic(self, processing_node=1, force=0):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
658 659
      """
        Starts again an activity
Jean-Paul Smets's avatar
Jean-Paul Smets committed
660
        processing_node starts from 1 (there is not node 0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
661
      """
Vincent Pelletier's avatar
Vincent Pelletier committed
662
      global active_threads, first_run
Jean-Paul Smets's avatar
Jean-Paul Smets committed
663 664

      # return if the number of threads is too high
665
      # else, increase the number of active_threads and continue
666 667
      tic_lock.acquire()
      too_many_threads = (active_threads >= max_active_threads)
668
      if not too_many_threads or force:
669
        active_threads += 1
670 671 672
      else:
        tic_lock.release()
        raise RuntimeError, 'Too many threads'
673
      tic_lock.release()
674

Jean-Paul Smets's avatar
Jean-Paul Smets committed
675
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
676 677
      if not is_initialized:
        self.initialize()
678

679
      inner_self = aq_inner(self)
680

681 682 683
      # If this is the first tic after zope is started, reset the processing
      # flag for activities of this node
      if first_run:
684 685 686 687
        inner_self.SQLDict_clearProcessingFlag(
                                processing_node=processing_node)
        inner_self.SQLQueue_clearProcessingFlag(
                                processing_node=processing_node)
688
        first_run = False
689

690
      try:
691 692 693 694 695 696 697
        #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)
        
698
        # Wakeup each queue
699
        for activity in activity_list:
700
          activity.wakeup(inner_self, processing_node)
701

702 703 704 705
        # Process messages on each queue in round robin
        has_awake_activity = 1
        while has_awake_activity:
          has_awake_activity = 0
706
          for activity in activity_list:
707 708
            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)
709 710 711 712 713
      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
714

715
    def hasActivity(self, *args, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
716
      # Check in each queue if the object has deferred tasks
717 718
      # if not argument is provided, then check on self
      if len(args) > 0:
719
        obj = args[0]
720
      else:
721
        obj = self
722
      for activity in activity_dict.itervalues():
723
        if activity.hasActivity(aq_inner(self), obj, **kw):
724 725
          return True
      return False
Jean-Paul Smets's avatar
Jean-Paul Smets committed
726

727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742
    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
      """
743 744
      # Safeguard: make sure we are wrapped in  acquisition context before
      # using our path as an activity tool instance-wide identifier.
745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763
      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

764 765
    security.declarePrivate('activateObject')
    def activateObject(self, object, activity, active_process, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
766 767
      if not is_initialized:
        self.initialize()
768
      self.getActivityBuffer()
769
      return ActiveWrapper(object, activity, active_process, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
770

771
    def deferredQueueMessage(self, activity, message):
772 773
      activity_buffer = self.getActivityBuffer()
      activity_buffer.deferredQueueMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
774

775
    def deferredDeleteMessage(self, activity, message):
776 777
      activity_buffer = self.getActivityBuffer()
      activity_buffer.deferredDeleteMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
778

Jean-Paul Smets's avatar
Jean-Paul Smets committed
779
    def getRegisteredMessageList(self, activity):
780
      activity_buffer = self.getActivityBuffer(create_if_not_found=False)
781
      if activity_buffer is not None:
782 783
        #activity_buffer._register() # This is required if flush flush is called outside activate
        return activity.getRegisteredMessageList(activity_buffer,
784
                                                 aq_inner(self))
785 786
      else:
        return []
Yoshinori Okuji's avatar
Yoshinori Okuji committed
787

Jean-Paul Smets's avatar
Jean-Paul Smets committed
788
    def unregisterMessage(self, activity, message):
789 790 791
      activity_buffer = self.getActivityBuffer()
      #activity_buffer._register()
      return activity.unregisterMessage(activity_buffer, aq_inner(self), message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
792

793
    def flush(self, obj, invoke=0, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
794 795
      if not is_initialized:
        self.initialize()
796
      self.getActivityBuffer()
797 798
      if isinstance(obj, tuple):
        object_path = obj
799
      else:
800
        object_path = obj.getPhysicalPath()
801
      for activity in activity_dict.itervalues():
802
        activity.flush(aq_inner(self), object_path, invoke=invoke, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
803

804
    def start(self, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
805 806
      if not is_initialized:
        self.initialize()
807
      for activity in activity_dict.itervalues():
808
        activity.start(aq_inner(self), **kw)
809 810

    def stop(self, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
811 812
      if not is_initialized:
        self.initialize()
813
      for activity in activity_dict.itervalues():
814
        activity.stop(aq_inner(self), **kw)
815

Jean-Paul Smets's avatar
Jean-Paul Smets committed
816
    def invoke(self, message):
817
      if logging:
818
        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))
819
      old_ihotfix_context = False
820 821
      if getattr(self, 'aq_chain', None) is not None:
        # Grab existing acquisition chain and extrach base objects.
822
        base_chain = [aq_base(x) for x in self.aq_chain]
823 824 825 826 827 828 829
        # 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:
830
          warn('CMFActivity.ActivityTool.invoke: PARENTS is not defined in REQUEST. It should only happen in unit tests.')
831
          request['PARENTS'] = self.aq_chain[:]
832 833 834
        # XXX: itools (used by iHotfix) requires PATH_INFO to be set, and it's
        # not when runing unit tests. Recreate it if it does not exist.
        if request.environ.get('PATH_INFO') is None:
835
          request.environ['PATH_INFO'] = '/Control_Panel/timer_service/process_timer'
836 837 838
        
        # restore request information
        new_request = request.clone()
839 840 841
        request_info = message.request_info
        new_request._script = request_info['_script']
        if 'SERVER_URL' in request_info:
842
          new_request.other['SERVER_URL'] = request_info['SERVER_URL']
843 844 845
        if 'VirtualRootPhysicalPath' in request_info:
          new_request.other['VirtualRootPhysicalPath'] = request_info['VirtualRootPhysicalPath']
        if 'HTTP_ACCEPT_LANGUAGE' in request_info:
846
          new_request.environ['HTTP_ACCEPT_LANGUAGE'] = request_info['HTTP_ACCEPT_LANGUAGE']
847 848 849 850 851 852 853 854 855 856 857
          # Replace iHotfix Context, saving existing one
          ihotfix_context = iHotfix.Context(new_request)
          id = get_ident()
          iHotfix._the_lock.acquire()
          try:
            old_ihotfix_context = iHotfix.contexts.get(id)
            iHotfix.contexts[id] = ihotfix_context
          finally:
            iHotfix._the_lock.release()
          # Execute iHotfix "patch 2"
          new_request.processInputs()
858 859

        new_request_container = request_container.__class__(REQUEST=new_request)
860 861 862 863 864 865 866 867
        # 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.')
868 869 870 871 872 873 874 875 876 877 878 879 880 881
      try:
        message(my_self)
      finally:
        if old_ihotfix_context is not False:
          # Restore iHotfix context
          id = get_ident()
          iHotfix._the_lock.acquire()
          try:
            if old_ihotfix_context is None:
              del iHotfix.contexts[id]
            else:
              iHotfix.contexts[id] = old_ihotfix_context
          finally:
            iHotfix._the_lock.release()
882
      if logging:
883
        LOG('Activity Tracking', INFO, 'invoked message')
884 885 886
      if my_self is not self: # We rewrapped self
        for held in my_self.REQUEST._held:
          self.REQUEST._hold(held)
887

888
    def invokeGroup(self, method_id, message_list):
889
      if logging:
890
        LOG('Activity Tracking', INFO, 'invoking group messages: method_id=%s, paths=%s' % (method_id, ['/'.join(m.object_path) for m in message_list]))
891 892 893 894 895
      # Invoke a group method.
      object_list = []
      expanded_object_list = []
      new_message_list = []
      path_dict = {}
896
      # Filter the list of messages. If an object is not available, mark its message as non-executable.
897 898
      # In addition, expand an object if necessary, and make sure that no duplication happens.
      for m in message_list:
899 900
        # alternate method is used to segregate objects which cannot be grouped.
        alternate_method_id = m.activity_kw.get('alternate_method_id')
901 902
        try:
          obj = m.getObject(self)
903 904 905 906
        except KeyError:
          m.is_executed = MESSAGE_NOT_EXECUTABLE
          continue
        try:
907
          i = len(new_message_list) # This is an index of this message in new_message_list.
908
          if m.hasExpandMethod():
909 910
            for subobj in m.getObjectList(self):
              path = subobj.getPath()
911
              if path not in path_dict:
912
                path_dict[path] = i
913 914 915 916 917 918
                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']
919 920
                  if 'group_id' in activity_kw:
                    del activity_kw['group_id']                    
921 922 923 924
                  active_obj = subobj.activate(**activity_kw)
                  getattr(active_obj, alternate_method_id)(*m.args, **m.kw)
                else:
                  expanded_object_list.append(subobj)
925 926 927
          else:
            path = obj.getPath()
            if path not in path_dict:
928
              path_dict[path] = i
929 930 931 932 933 934
              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']
935 936
                if 'group_id' in activity_kw:
                  del activity_kw['group_id']
937 938 939 940
                active_obj = obj.activate(**activity_kw)
                getattr(active_obj, alternate_method_id)(*m.args, **m.kw)
              else:
                expanded_object_list.append(obj)
941
          object_list.append(obj)
942 943
          new_message_list.append(m)
        except:
944
          m.is_executed = MESSAGE_NOT_EXECUTED
945 946 947 948 949
          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))
950
          LOG('WARNING ActivityTool', 0,
951
              'Could not call method %s on object %s' %
952
              (m.method_id, m.object_path), error=exc_info)
953

954 955
      try:
        if len(expanded_object_list) > 0:
956 957
          method = self.unrestrictedTraverse(method_id)
          # FIXME: how to apply security here?
958 959
          # 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.
960
          result = method(expanded_object_list, **m.kw)
961
        else:
962 963 964
          result = None
      except:
        # In this case, the group method completely failed.
965 966 967 968 969
        exc_info = sys.exc_info()
        exc_type = exc_info[0]
        exc_value = str(exc_info[1])
        traceback = ''.join(ExceptionFormatter.format_exception(
                            *exc_info))
970
        for m in new_message_list:
971
          m.is_executed = MESSAGE_NOT_EXECUTED
972 973 974
          m.exc_type = exc_type
          m.exc_value = exc_value
          m.traceback = traceback
975
        LOG('WARNING ActivityTool', 0,
976
            'Could not call method %s on objects %s' %
977
            (method_id, expanded_object_list), error=exc_info)
978 979 980 981 982 983 984 985 986 987 988 989 990
      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:
991
            m.is_executed = MESSAGE_NOT_EXECUTED
992
            LOG('ActivityTool', WARNING,
993 994
                'the method %s partially failed on object %s' %
                (m.method_id, m.object_path,))
995 996 997
          else:
            try:
              m.activateResult(self, result, object)
998
              m.is_executed = MESSAGE_EXECUTED
999
            except:
1000
              m.is_executed = MESSAGE_NOT_EXECUTED
Vincent Pelletier's avatar
Vincent Pelletier committed
1001
              m.exc_type = sys.exc_info()[0]
1002
              LOG('ActivityTool', WARNING,
1003
                  'Could not call method %s on object %s' % (
Vincent Pelletier's avatar
Vincent Pelletier committed
1004
                  m.method_id, m.object_path), error=sys.exc_info())
1005
      if logging:
1006
        LOG('Activity Tracking', INFO, 'invoked group messages')
1007

1008 1009
    def newMessage(self, activity, path, active_process,
                   activity_kw, method_id, *args, **kw):
1010
      # Some Security Cheking should be made here XXX
Vincent Pelletier's avatar
Vincent Pelletier committed
1011 1012
      if not is_initialized:
        self.initialize()
1013
      self.getActivityBuffer()
1014
      activity_dict[activity].queueMessage(aq_inner(self),
1015
        Message(path, active_process, activity_kw, method_id, args, kw))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1016

1017
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageInvoke' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1018 1019 1020 1021 1022 1023
    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('/'))
1024
      self.flush(object_path,method_id=method_id,invoke=1)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1025
      if REQUEST is not None:
1026 1027
        return REQUEST.RESPONSE.redirect('%s/%s' %
                (self.absolute_url(), 'manageActivities'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1028

1029
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageCancel' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1030 1031 1032 1033 1034 1035
    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('/'))
1036
      self.flush(object_path,method_id=method_id,invoke=0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1037
      if REQUEST is not None:
1038 1039
        return REQUEST.RESPONSE.redirect('%s/%s' %
                (self.absolute_url(), 'manageActivities'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1040

1041 1042
    security.declareProtected( CMFCorePermissions.ManagePortal,
                               'manageClearActivities' )
1043
    def manageClearActivities(self, keep=1, REQUEST=None):
1044 1045 1046 1047 1048
      """
        Clear all activities and recreate tables.
      """
      folder = getToolByName(self, 'portal_skins').activity

1049 1050
      # Obtain all pending messages.
      message_list = []
1051
      if keep:
1052
        for activity in activity_dict.itervalues():
1053 1054 1055 1056 1057 1058
          if hasattr(activity, 'dumpMessageList'):
            try:
              message_list.extend(activity.dumpMessageList(self))
            except ConflictError:
              raise
            except:
1059 1060 1061
              LOG('ActivityTool', WARNING,
                  'could not dump messages from %s' %
                  (activity,), error=sys.exc_info())
1062 1063

      if getattr(folder, 'SQLDict_createMessageTable', None) is not None:
1064 1065 1066 1067 1068
        try:
          folder.SQLDict_dropMessageTable()
        except ConflictError:
          raise
        except:
1069
          LOG('CMFActivity', WARNING,
1070
              'could not drop the message table',
1071 1072 1073
              error=sys.exc_info())
        folder.SQLDict_createMessageTable()

1074
      if getattr(folder, 'SQLQueue_createMessageTable', None) is not None:
1075 1076 1077 1078 1079
        try:
          folder.SQLQueue_dropMessageTable()
        except ConflictError:
          raise
        except:
1080
          LOG('CMFActivity', WARNING,
1081
              'could not drop the message queue table',
1082 1083 1084
              error=sys.exc_info())
        folder.SQLQueue_createMessageTable()

1085 1086 1087
      # Reactivate the messages.
      for m in message_list:
        try:
1088
          m.reactivate(aq_inner(self))
1089 1090 1091 1092
        except ConflictError:
          raise
        except:
          LOG('ActivityTool', WARNING,
1093 1094
              'could not reactivate the message %r, %r' %
              (m.object_path, m.method_id), error=sys.exc_info())
1095

1096
      if REQUEST is not None:
1097 1098 1099 1100 1101 1102
        message = 'Activities%20Cleared'
        if keep:
          message = 'Tables%20Recreated'
        return REQUEST.RESPONSE.redirect(
            '%s/manageActivitiesAdvanced?manage_tabs_message=%s' % (
              self.absolute_url(), message))
1103

Jean-Paul Smets's avatar
Jean-Paul Smets committed
1104
    security.declarePublic('getMessageList')
1105
    def getMessageList(self,**kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1106 1107 1108
      """
        List messages waiting in queues
      """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1109
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
1110 1111
      if not is_initialized:
        self.initialize()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1112

Jean-Paul Smets's avatar
Jean-Paul Smets committed
1113
      message_list = []
1114
      for activity in activity_dict.itervalues():
Sebastien Robin's avatar
Sebastien Robin committed
1115
        try:
1116
          message_list += activity.getMessageList(aq_inner(self),**kw)
Sebastien Robin's avatar
Sebastien Robin committed
1117 1118
        except AttributeError:
          LOG('getMessageList, could not get message from Activity:',0,activity)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1119 1120
      return message_list

1121 1122 1123 1124 1125 1126
    security.declarePublic('countMessageWithTag')
    def countMessageWithTag(self, value):
      """
        Return the number of messages which match the given tag.
      """
      message_count = 0
1127
      for activity in activity_dict.itervalues():
1128
        message_count += activity.countMessageWithTag(aq_inner(self), value)
Sebastien Robin's avatar
Sebastien Robin committed
1129 1130 1131 1132 1133 1134 1135 1136 1137 1138
      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
1139
        path : for activities on a particular object
Sebastien Robin's avatar
Sebastien Robin committed
1140 1141 1142 1143
        tag : activities with a particular tag
        message_uid : activities with a particular uid
      """
      message_count = 0
1144
      for activity in activity_dict.itervalues():
1145
        message_count += activity.countMessage(aq_inner(self), **kw)
1146 1147
      return message_count

1148
    security.declareProtected( CMFCorePermissions.ManagePortal , 'newActiveProcess' )
1149
    def newActiveProcess(self, **kw):
1150 1151 1152
      from ActiveProcess import addActiveProcess
      new_id = str(self.generateNewId())
      addActiveProcess(self, new_id)
1153 1154 1155
      active_process = self._getOb(new_id)
      active_process.edit(**kw)
      return active_process
1156 1157 1158 1159

    def reindexObject(self):
      self.immediateReindexObject()

1160
    # Active synchronisation methods
1161
    security.declarePrivate('validateOrder')
1162
    def validateOrder(self, message, validator_id, validation_value):
1163 1164 1165 1166 1167
      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
1168 1169
      if not is_initialized:
        self.initialize()
1170
      message_list = []
Vincent Pelletier's avatar
Vincent Pelletier committed
1171
      method_id = "_validate_%s" % validator_id
1172
      for activity in activity_dict.itervalues():
1173 1174 1175 1176 1177 1178
        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
1179

Yoshinori Okuji's avatar
Yoshinori Okuji committed
1180 1181
    # Required for tests (time shift)
    def timeShift(self, delay):
Vincent Pelletier's avatar
Vincent Pelletier committed
1182 1183
      if not is_initialized:
        self.initialize()
1184
      for activity in activity_dict.itervalues():
1185
        activity.timeShift(aq_inner(self), delay)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
1186

1187
InitializeClass(ActivityTool)