ActivityTool.py 48 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
is_running_lock = threading.Lock()
77
first_run = True
78 79 80
currentNode = None
ROLE_IDLE = 0
ROLE_PROCESSING = 1
Jean-Paul Smets's avatar
Jean-Paul Smets committed
81 82 83 84

# Activity Registration
activity_dict = {}

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

def enableLogging():
  global logging
  logging = True

def disableLogging():
  global logging
  logging = False

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

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

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

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

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

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

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

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

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

Jean-Paul Smets's avatar
Jean-Paul Smets committed
218
  def __call__(self, activity_tool):
219
    try:
220
      obj = self.getObject(activity_tool)
221 222 223
    except KeyError:
      self.is_executed = MESSAGE_NOT_EXECUTABLE
    else:
224
      try:
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 255
        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
256

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

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

%s

Document: %s
Method: %s
278 279
Arguments:%r
Named Parameters:%r
Vincent Pelletier's avatar
Vincent Pelletier committed
280
Exception: %s %s
281

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

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

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

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

321 322
allow_class(Method)

Jean-Paul Smets's avatar
Jean-Paul Smets committed
323 324
class ActiveWrapper:

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

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

336 337 338 339 340 341 342
# Set to False when shutting down. Access outside of process_shutdown must
# be done under the protection of is_running_lock lock.
is_running = True
# True when activities cannot be executing any more.
has_processed_shutdown = False


343
class ActivityTool (Folder, UniqueObject):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
344
    """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
345 346 347 348 349 350 351 352 353 354 355 356
    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
357 358 359
    """
    id = 'portal_activities'
    meta_type = 'CMF Activity Tool'
360
    portal_type = 'Activity Tool'
361
    allowed_types = ( 'CMF Active Process', )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
362 363
    security = ClassSecurityInfo()

364 365
    manage_options = tuple(
                     [ { 'label' : 'Overview', 'action' : 'manage_overview' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
366
                     , { 'label' : 'Activities', 'action' : 'manageActivities' }
367
                     , { 'label' : 'LoadBalancing', 'action' : 'manageLoadBalancing'}
368
                     , { 'label' : 'Advanced', 'action' : 'manageActivitiesAdvanced' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
369
                     ,
370
                     ] + list(Folder.manage_options))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
371 372 373 374

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

375 376 377
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageActivitiesAdvanced' )
    manageActivitiesAdvanced = DTMLFile( 'dtml/manageActivitiesAdvanced', globals() )

378 379
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manage_overview' )
    manage_overview = DTMLFile( 'dtml/explainActivityTool', globals() )
380 381 382 383 384 385
    
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageLoadBalancing' )
    manageLoadBalancing = DTMLFile( 'dtml/manageLoadBalancing', globals() )
    
    distributingNode = ''
    _nodes = ()
386 387 388

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

390 391 392 393 394 395 396 397 398 399
    # 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
400 401
    def initialize(self):
      global is_initialized
Sebastien Robin's avatar
Sebastien Robin committed
402
      from Activity import RAMQueue, RAMDict, SQLQueue, SQLDict
Jean-Paul Smets's avatar
Jean-Paul Smets committed
403
      # Initialize each queue
404
      for activity in activity_dict.itervalues():
Jean-Paul Smets's avatar
Jean-Paul Smets committed
405
        activity.initialize(self)
Vincent Pelletier's avatar
Vincent Pelletier committed
406
      is_initialized = True
407 408 409
      
    security.declareProtected(Permissions.manage_properties, 'isSubscribed')
    def isSubscribed(self):
Aurel's avatar
Aurel committed
410
        """
411 412 413 414 415 416 417 418 419 420 421 422
        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
423

424
    security.declareProtected(Permissions.manage_properties, 'subscribe')
425
    def subscribe(self, REQUEST=None, RESPONSE=None):
426 427
        """ subscribe to the global Timer Service """
        service = getTimerService(self)
428
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
429
        if not service:
430
            LOG('ActivityTool', INFO, 'TimerService not available')
431 432 433 434
            url += urllib.quote('TimerService not available')
        else:
            service.subscribe(self)
            url += urllib.quote("Subscribed to Timer Service")
435 436
        if RESPONSE is not None:
            RESPONSE.redirect(url)
437 438

    security.declareProtected(Permissions.manage_properties, 'unsubscribe')
439
    def unsubscribe(self, REQUEST=None, RESPONSE=None):
440 441
        """ unsubscribe from the global Timer Service """
        service = getTimerService(self)
442
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
443
        if not service:
444
            LOG('ActivityTool', INFO, 'TimerService not available')
445 446 447 448
            url += urllib.quote('TimerService not available')
        else:
            service.unsubscribe(self)
            url += urllib.quote("Unsubscribed from Timer Service")
449 450
        if RESPONSE is not None:
            RESPONSE.redirect(url)
451 452 453

    def manage_beforeDelete(self, item, container):
        self.unsubscribe()
454 455
        Folder.inheritedAttribute('manage_beforeDelete')(self, item, container)
    
456 457
    def manage_afterAdd(self, item, container):
        self.subscribe()
458 459
        Folder.inheritedAttribute('manage_afterAdd')(self, item, container)
       
460 461
    def getCurrentNode(self):
        """ Return current node in form ip:port """
462 463 464 465 466 467 468 469 470 471 472 473 474
        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)
475 476 477 478 479 480 481
        return currentNode
        
    security.declarePublic('getDistributingNode')
    def getDistributingNode(self):
        """ Return the distributingNode """
        return self.distributingNode

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 509 510 511 512 513 514 515 516 517 518
    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)

519
    security.declareProtected(CMFCorePermissions.ManagePortal, 'getIdleNodeList')
520 521
    def getIdleNodeList(self):
      return self.getNodeList(role=ROLE_IDLE)
522

523 524 525 526
    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)
      
527 528
    security.declarePublic('manage_setDistributingNode')
    def manage_setDistributingNode(self, distributingNode, REQUEST=None):
529
        """ set the distributing node """   
530
        if not distributingNode or self._isValidNodeName(distributingNode):
531 532 533 534 535 536 537 538 539 540 541 542 543
          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."))

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 592 593 594 595 596 597 598 599 600 601
    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))
602

603 604 605 606 607
    def process_shutdown(self, phase, time_in_phase):
        """
          Prevent shutdown from happening while an activity queue is
          processing a batch.
        """
608 609 610
        is_running = False
        if phase == 3 and not has_processed_shutdown:
          has_processed_shutdown = True
611 612 613 614 615
          LOG('CMFActivity', INFO, "Shutdown: Waiting for activities to finish.")
          is_running_lock.acquire()
          LOG('CMFActivity', INFO, "Shutdown: Activities finished.")
          is_running_lock.release()

616
    def process_timer(self, tick, interval, prev="", next=""):
617
        """
618 619 620 621 622
        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.
        """
623 624 625 626
        # Prevent TimerService from starting multiple threads in parallel
        acquired = timerservice_lock.acquire(0)
        if not acquired:
          return
627

628
        try:
629
          old_sm = getSecurityManager()
630
          try:
631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662
            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
663
        finally:
664
          timerservice_lock.release()
665

Jean-Paul Smets's avatar
Jean-Paul Smets committed
666 667 668 669 670 671
    security.declarePublic('distribute')
    def distribute(self, node_count=1):
      """
        Distribute load
      """
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
672 673
      if not is_initialized:
        self.initialize()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
674 675

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

Jean-Paul Smets's avatar
Jean-Paul Smets committed
679
    security.declarePublic('tic')
Jean-Paul Smets's avatar
Jean-Paul Smets committed
680
    def tic(self, processing_node=1, force=0):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
681 682
      """
        Starts again an activity
Jean-Paul Smets's avatar
Jean-Paul Smets committed
683
        processing_node starts from 1 (there is not node 0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
684
      """
Vincent Pelletier's avatar
Vincent Pelletier committed
685
      global active_threads, first_run
Jean-Paul Smets's avatar
Jean-Paul Smets committed
686 687

      # return if the number of threads is too high
688
      # else, increase the number of active_threads and continue
689 690
      tic_lock.acquire()
      too_many_threads = (active_threads >= max_active_threads)
691
      if not too_many_threads or force:
692
        active_threads += 1
693 694 695
      else:
        tic_lock.release()
        raise RuntimeError, 'Too many threads'
696
      tic_lock.release()
697

Jean-Paul Smets's avatar
Jean-Paul Smets committed
698
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
699 700
      if not is_initialized:
        self.initialize()
701

702
      inner_self = aq_inner(self)
703

704 705 706
      # If this is the first tic after zope is started, reset the processing
      # flag for activities of this node
      if first_run:
707 708 709 710
        inner_self.SQLDict_clearProcessingFlag(
                                processing_node=processing_node)
        inner_self.SQLQueue_clearProcessingFlag(
                                processing_node=processing_node)
711
        first_run = False
712

713
      try:
714 715 716 717 718 719 720
        #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)
        
721
        # Wakeup each queue
722
        for activity in activity_list:
723
          activity.wakeup(inner_self, processing_node)
724

725 726 727 728
        # Process messages on each queue in round robin
        has_awake_activity = 1
        while has_awake_activity:
          has_awake_activity = 0
729
          for activity in activity_list:
730 731
            is_running_lock.acquire()
            try:
732
              if is_running:
733 734 735 736
                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)
            finally:
              is_running_lock.release()
737 738 739 740 741
      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
742

743
    def hasActivity(self, *args, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
744
      # Check in each queue if the object has deferred tasks
745 746
      # if not argument is provided, then check on self
      if len(args) > 0:
747
        obj = args[0]
748
      else:
749
        obj = self
750
      for activity in activity_dict.itervalues():
751
        if activity.hasActivity(aq_inner(self), obj, **kw):
752 753
          return True
      return False
Jean-Paul Smets's avatar
Jean-Paul Smets committed
754

755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770
    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
      """
771 772
      # Safeguard: make sure we are wrapped in  acquisition context before
      # using our path as an activity tool instance-wide identifier.
773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791
      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

792 793
    security.declarePrivate('activateObject')
    def activateObject(self, object, activity, active_process, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
794 795
      if not is_initialized:
        self.initialize()
796
      self.getActivityBuffer()
797
      return ActiveWrapper(object, activity, active_process, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
798

799
    def deferredQueueMessage(self, activity, message):
800 801
      activity_buffer = self.getActivityBuffer()
      activity_buffer.deferredQueueMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
802

803
    def deferredDeleteMessage(self, activity, message):
804 805
      activity_buffer = self.getActivityBuffer()
      activity_buffer.deferredDeleteMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
806

Jean-Paul Smets's avatar
Jean-Paul Smets committed
807
    def getRegisteredMessageList(self, activity):
808
      activity_buffer = self.getActivityBuffer(create_if_not_found=False)
809
      if activity_buffer is not None:
810 811
        #activity_buffer._register() # This is required if flush flush is called outside activate
        return activity.getRegisteredMessageList(activity_buffer,
812
                                                 aq_inner(self))
813 814
      else:
        return []
Yoshinori Okuji's avatar
Yoshinori Okuji committed
815

Jean-Paul Smets's avatar
Jean-Paul Smets committed
816
    def unregisterMessage(self, activity, message):
817 818 819
      activity_buffer = self.getActivityBuffer()
      #activity_buffer._register()
      return activity.unregisterMessage(activity_buffer, aq_inner(self), message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
820

821
    def flush(self, obj, invoke=0, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
822 823
      if not is_initialized:
        self.initialize()
824
      self.getActivityBuffer()
825 826
      if isinstance(obj, tuple):
        object_path = obj
827
      else:
828
        object_path = obj.getPhysicalPath()
829
      for activity in activity_dict.itervalues():
830
        activity.flush(aq_inner(self), object_path, invoke=invoke, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
831

832
    def start(self, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
833 834
      if not is_initialized:
        self.initialize()
835
      for activity in activity_dict.itervalues():
836
        activity.start(aq_inner(self), **kw)
837 838

    def stop(self, **kw):
Vincent Pelletier's avatar
Vincent Pelletier committed
839 840
      if not is_initialized:
        self.initialize()
841
      for activity in activity_dict.itervalues():
842
        activity.stop(aq_inner(self), **kw)
843

Jean-Paul Smets's avatar
Jean-Paul Smets committed
844
    def invoke(self, message):
845
      if logging:
846
        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))
847
      old_ihotfix_context = False
848 849
      if getattr(self, 'aq_chain', None) is not None:
        # Grab existing acquisition chain and extrach base objects.
850
        base_chain = [aq_base(x) for x in self.aq_chain]
851 852 853
        # Grab existig request (last chain item) and create a copy.
        request_container = base_chain.pop()
        request = request_container.REQUEST
854 855 856 857 858 859 860 861
        # Generate PARENTS value. Sadly, we cannot reuse base_chain since
        # PARENTS items must be wrapped in acquisition
        parents = []
        application = self.getPhysicalRoot().aq_base
        for parent in self.aq_chain:
          if parent.aq_base is application:
            break
          parents.append(parent)
862 863
        # XXX: REQUEST.clone() requires PARENTS to be set, and it's not when
        # runing unit tests. Recreate it if it does not exist.
864 865
        if getattr(request.other, 'PARENTS', None) is None:
          request.other['PARENTS'] = parents
866 867 868
        # 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:
869
          request.environ['PATH_INFO'] = '/Control_Panel/timer_service/process_timer'
870 871 872
        
        # restore request information
        new_request = request.clone()
873
        request_info = message.request_info
874 875
        # PARENTS is truncated by clone
        new_request.other['PARENTS'] = parents
876 877
        new_request._script = request_info['_script']
        if 'SERVER_URL' in request_info:
878
          new_request.other['SERVER_URL'] = request_info['SERVER_URL']
879 880 881
        if 'VirtualRootPhysicalPath' in request_info:
          new_request.other['VirtualRootPhysicalPath'] = request_info['VirtualRootPhysicalPath']
        if 'HTTP_ACCEPT_LANGUAGE' in request_info:
882
          new_request.environ['HTTP_ACCEPT_LANGUAGE'] = request_info['HTTP_ACCEPT_LANGUAGE']
883 884 885 886 887 888 889 890 891 892 893
          # 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()
894 895

        new_request_container = request_container.__class__(REQUEST=new_request)
896 897 898 899 900 901 902 903
        # 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.')
904 905 906 907 908 909 910 911 912 913 914 915 916 917
      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()
918
      if logging:
919
        LOG('Activity Tracking', INFO, 'invoked message')
920 921 922
      if my_self is not self: # We rewrapped self
        for held in my_self.REQUEST._held:
          self.REQUEST._hold(held)
923

924
    def invokeGroup(self, method_id, message_list):
925
      if logging:
926
        LOG('Activity Tracking', INFO, 'invoking group messages: method_id=%s, paths=%s' % (method_id, ['/'.join(m.object_path) for m in message_list]))
927 928 929 930 931
      # Invoke a group method.
      object_list = []
      expanded_object_list = []
      new_message_list = []
      path_dict = {}
932
      # Filter the list of messages. If an object is not available, mark its message as non-executable.
933 934
      # In addition, expand an object if necessary, and make sure that no duplication happens.
      for m in message_list:
935 936
        # alternate method is used to segregate objects which cannot be grouped.
        alternate_method_id = m.activity_kw.get('alternate_method_id')
937 938
        try:
          obj = m.getObject(self)
939 940 941 942
        except KeyError:
          m.is_executed = MESSAGE_NOT_EXECUTABLE
          continue
        try:
943
          i = len(new_message_list) # This is an index of this message in new_message_list.
944
          if m.hasExpandMethod():
945 946
            for subobj in m.getObjectList(self):
              path = subobj.getPath()
947
              if path not in path_dict:
948
                path_dict[path] = i
949 950 951 952 953 954
                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']
955 956
                  if 'group_id' in activity_kw:
                    del activity_kw['group_id']                    
957 958 959 960
                  active_obj = subobj.activate(**activity_kw)
                  getattr(active_obj, alternate_method_id)(*m.args, **m.kw)
                else:
                  expanded_object_list.append(subobj)
961 962 963
          else:
            path = obj.getPath()
            if path not in path_dict:
964
              path_dict[path] = i
965 966 967 968 969 970
              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']
971 972
                if 'group_id' in activity_kw:
                  del activity_kw['group_id']
973 974 975 976
                active_obj = obj.activate(**activity_kw)
                getattr(active_obj, alternate_method_id)(*m.args, **m.kw)
              else:
                expanded_object_list.append(obj)
977
          object_list.append(obj)
978 979
          new_message_list.append(m)
        except:
980
          m.is_executed = MESSAGE_NOT_EXECUTED
981 982 983 984 985
          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))
986
          LOG('WARNING ActivityTool', 0,
987
              'Could not call method %s on object %s' %
988
              (m.method_id, m.object_path), error=exc_info)
989

990 991
      try:
        if len(expanded_object_list) > 0:
992 993
          method = self.unrestrictedTraverse(method_id)
          # FIXME: how to apply security here?
994 995
          # 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.
996
          result = method(expanded_object_list, **m.kw)
997
        else:
998 999 1000
          result = None
      except:
        # In this case, the group method completely failed.
1001 1002 1003 1004 1005
        exc_info = sys.exc_info()
        exc_type = exc_info[0]
        exc_value = str(exc_info[1])
        traceback = ''.join(ExceptionFormatter.format_exception(
                            *exc_info))
1006
        for m in new_message_list:
1007
          m.is_executed = MESSAGE_NOT_EXECUTED
1008 1009 1010
          m.exc_type = exc_type
          m.exc_value = exc_value
          m.traceback = traceback
1011
        LOG('WARNING ActivityTool', 0,
1012
            'Could not call method %s on objects %s' %
1013
            (method_id, expanded_object_list), error=exc_info)
1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026
      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:
1027
            m.is_executed = MESSAGE_NOT_EXECUTED
1028
            LOG('ActivityTool', WARNING,
1029 1030
                'the method %s partially failed on object %s' %
                (m.method_id, m.object_path,))
1031 1032 1033
          else:
            try:
              m.activateResult(self, result, object)
1034
              m.is_executed = MESSAGE_EXECUTED
1035
            except:
1036
              m.is_executed = MESSAGE_NOT_EXECUTED
Vincent Pelletier's avatar
Vincent Pelletier committed
1037
              m.exc_type = sys.exc_info()[0]
1038
              LOG('ActivityTool', WARNING,
1039
                  'Could not call method %s on object %s' % (
Vincent Pelletier's avatar
Vincent Pelletier committed
1040
                  m.method_id, m.object_path), error=sys.exc_info())
1041
      if logging:
1042
        LOG('Activity Tracking', INFO, 'invoked group messages')
1043

1044 1045
    def newMessage(self, activity, path, active_process,
                   activity_kw, method_id, *args, **kw):
1046
      # Some Security Cheking should be made here XXX
Vincent Pelletier's avatar
Vincent Pelletier committed
1047 1048
      if not is_initialized:
        self.initialize()
1049
      self.getActivityBuffer()
1050
      activity_dict[activity].queueMessage(aq_inner(self),
1051
        Message(path, active_process, activity_kw, method_id, args, kw))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1052

1053
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageInvoke' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1054 1055 1056 1057 1058 1059
    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('/'))
1060
      self.flush(object_path,method_id=method_id,invoke=1)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1061
      if REQUEST is not None:
1062 1063
        return REQUEST.RESPONSE.redirect('%s/%s' %
                (self.absolute_url(), 'manageActivities'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1064

1065
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageCancel' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1066 1067 1068 1069 1070 1071
    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('/'))
1072
      self.flush(object_path,method_id=method_id,invoke=0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1073
      if REQUEST is not None:
1074 1075
        return REQUEST.RESPONSE.redirect('%s/%s' %
                (self.absolute_url(), 'manageActivities'))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1076

1077 1078
    security.declareProtected( CMFCorePermissions.ManagePortal,
                               'manageClearActivities' )
1079
    def manageClearActivities(self, keep=1, REQUEST=None):
1080 1081 1082 1083 1084
      """
        Clear all activities and recreate tables.
      """
      folder = getToolByName(self, 'portal_skins').activity

1085 1086
      # Obtain all pending messages.
      message_list = []
1087
      if keep:
1088
        for activity in activity_dict.itervalues():
1089 1090 1091 1092 1093 1094
          if hasattr(activity, 'dumpMessageList'):
            try:
              message_list.extend(activity.dumpMessageList(self))
            except ConflictError:
              raise
            except:
1095 1096 1097
              LOG('ActivityTool', WARNING,
                  'could not dump messages from %s' %
                  (activity,), error=sys.exc_info())
1098 1099

      if getattr(folder, 'SQLDict_createMessageTable', None) is not None:
1100 1101 1102 1103 1104
        try:
          folder.SQLDict_dropMessageTable()
        except ConflictError:
          raise
        except:
1105
          LOG('CMFActivity', WARNING,
1106
              'could not drop the message table',
1107 1108 1109
              error=sys.exc_info())
        folder.SQLDict_createMessageTable()

1110
      if getattr(folder, 'SQLQueue_createMessageTable', None) is not None:
1111 1112 1113 1114 1115
        try:
          folder.SQLQueue_dropMessageTable()
        except ConflictError:
          raise
        except:
1116
          LOG('CMFActivity', WARNING,
1117
              'could not drop the message queue table',
1118 1119 1120
              error=sys.exc_info())
        folder.SQLQueue_createMessageTable()

1121 1122 1123
      # Reactivate the messages.
      for m in message_list:
        try:
1124
          m.reactivate(aq_inner(self))
1125 1126 1127 1128
        except ConflictError:
          raise
        except:
          LOG('ActivityTool', WARNING,
1129 1130
              'could not reactivate the message %r, %r' %
              (m.object_path, m.method_id), error=sys.exc_info())
1131

1132
      if REQUEST is not None:
1133 1134 1135 1136 1137 1138
        message = 'Activities%20Cleared'
        if keep:
          message = 'Tables%20Recreated'
        return REQUEST.RESPONSE.redirect(
            '%s/manageActivitiesAdvanced?manage_tabs_message=%s' % (
              self.absolute_url(), message))
1139

Jean-Paul Smets's avatar
Jean-Paul Smets committed
1140
    security.declarePublic('getMessageList')
1141
    def getMessageList(self,**kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1142 1143 1144
      """
        List messages waiting in queues
      """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1145
      # Initialize if needed
Vincent Pelletier's avatar
Vincent Pelletier committed
1146 1147
      if not is_initialized:
        self.initialize()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1148

Jean-Paul Smets's avatar
Jean-Paul Smets committed
1149
      message_list = []
1150
      for activity in activity_dict.itervalues():
Sebastien Robin's avatar
Sebastien Robin committed
1151
        try:
1152
          message_list += activity.getMessageList(aq_inner(self),**kw)
Sebastien Robin's avatar
Sebastien Robin committed
1153 1154
        except AttributeError:
          LOG('getMessageList, could not get message from Activity:',0,activity)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1155 1156
      return message_list

1157 1158 1159 1160 1161 1162
    security.declarePublic('countMessageWithTag')
    def countMessageWithTag(self, value):
      """
        Return the number of messages which match the given tag.
      """
      message_count = 0
1163
      for activity in activity_dict.itervalues():
1164
        message_count += activity.countMessageWithTag(aq_inner(self), value)
Sebastien Robin's avatar
Sebastien Robin committed
1165 1166 1167 1168 1169 1170 1171 1172 1173 1174
      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
1175
        path : for activities on a particular object
Sebastien Robin's avatar
Sebastien Robin committed
1176 1177 1178 1179
        tag : activities with a particular tag
        message_uid : activities with a particular uid
      """
      message_count = 0
1180
      for activity in activity_dict.itervalues():
1181
        message_count += activity.countMessage(aq_inner(self), **kw)
1182 1183
      return message_count

1184
    security.declareProtected( CMFCorePermissions.ManagePortal , 'newActiveProcess' )
1185
    def newActiveProcess(self, **kw):
1186 1187 1188
      from ActiveProcess import addActiveProcess
      new_id = str(self.generateNewId())
      addActiveProcess(self, new_id)
1189 1190 1191
      active_process = self._getOb(new_id)
      active_process.edit(**kw)
      return active_process
1192 1193 1194 1195

    def reindexObject(self):
      self.immediateReindexObject()

1196
    # Active synchronisation methods
1197
    security.declarePrivate('validateOrder')
1198
    def validateOrder(self, message, validator_id, validation_value):
1199 1200 1201 1202 1203
      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
1204 1205
      if not is_initialized:
        self.initialize()
1206
      message_list = []
Vincent Pelletier's avatar
Vincent Pelletier committed
1207
      method_id = "_validate_%s" % validator_id
1208
      for activity in activity_dict.itervalues():
1209 1210 1211 1212 1213 1214
        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
1215

Yoshinori Okuji's avatar
Yoshinori Okuji committed
1216 1217
    # Required for tests (time shift)
    def timeShift(self, delay):
Vincent Pelletier's avatar
Vincent Pelletier committed
1218 1219
      if not is_initialized:
        self.initialize()
1220
      for activity in activity_dict.itervalues():
1221
        activity.timeShift(aq_inner(self), delay)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
1222

1223
InitializeClass(ActivityTool)