ActivityTool.py 31.4 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
import socket, asyncore, urllib
Jean-Paul Smets's avatar
Jean-Paul Smets committed
30
from Products.CMFCore import CMFCorePermissions
31
from Products.ERP5Type.Document.Folder import Folder
32 33
from Products.ERP5Type.Utils import getPath
from Products.ERP5Type.Error import Error
34
from Products.PythonScripts.Utility import allow_class
35 36 37
from App.ApplicationManager import ApplicationManager
from AccessControl import ClassSecurityInfo, Permissions
from AccessControl.SecurityManagement import newSecurityManager
38
from Products.CMFCore.utils import UniqueObject, _checkPermission, _getAuthenticatedUser, getToolByName
39
from Globals import InitializeClass, DTMLFile, get_request
Jean-Paul Smets's avatar
Jean-Paul Smets committed
40 41
from Acquisition import aq_base
from DateTime.DateTime import DateTime
42
from Products.CMFActivity.ActiveObject import DISTRIBUTABLE_STATE, INVOKE_ERROR_STATE, VALIDATE_ERROR_STATE
43
from ActivityBuffer import ActivityBuffer
44
from AccessControl.SecurityManagement import newSecurityManager
Jean-Paul Smets's avatar
Jean-Paul Smets committed
45
import threading
46
import sys
47
from ZODB.POSException import ConflictError
48
from OFS.Traversable import NotFound
Yoshinori Okuji's avatar
Yoshinori Okuji committed
49
from types import TupleType, StringType
Jean-Paul Smets's avatar
Jean-Paul Smets committed
50

51
from zLOG import LOG, INFO, WARNING
52 53

try:
54
  from Products.TimerService import getTimerService
55
except ImportError:
56 57
  def getTimerService(self):
    pass
Jean-Paul Smets's avatar
Jean-Paul Smets committed
58 59 60 61 62 63

# 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)
is_initialized = 0
64 65
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
66
first_run = 1
Jean-Paul Smets's avatar
Jean-Paul Smets committed
67 68 69 70 71 72 73 74

# Activity Registration
activity_dict = {}
activity_list = []

def registerActivity(activity):
  # Must be rewritten to register
  # class and create instance for each activity
75
  #LOG('Init Activity', 0, str(activity.__name__))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
76 77 78 79 80
  activity_instance = activity()
  activity_list.append(activity_instance)
  activity_dict[activity.__name__] = activity_instance

class Message:
Yoshinori Okuji's avatar
Yoshinori Okuji committed
81

82
  def __init__(self, object, active_process, activity_kw, method_id, args, kw):
Yoshinori Okuji's avatar
Yoshinori Okuji committed
83
    if type(object) is StringType:
Jean-Paul Smets's avatar
Jean-Paul Smets committed
84 85 86
      self.object_path = object.split('/')
    else:
      self.object_path = object.getPhysicalPath()
Yoshinori Okuji's avatar
Yoshinori Okuji committed
87
    if type(active_process) is StringType:
88 89 90 91 92
      self.active_process = active_process.split('/')
    elif active_process is None:
      self.active_process = None
    else:
      self.active_process = active_process.getPhysicalPath()
93
      self.active_process_uid = active_process.getUid()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
94 95 96 97
    self.activity_kw = activity_kw
    self.method_id = method_id
    self.args = args
    self.kw = kw
Jean-Paul Smets's avatar
Jean-Paul Smets committed
98
    self.is_executed = 0
99
    self.exc_type = None
100 101
    self.user_name = str(_getAuthenticatedUser(self))
    # Store REQUEST Info ?
Jean-Paul Smets's avatar
Jean-Paul Smets committed
102

103 104 105 106
  def getObject(self, activity_tool):
    return activity_tool.unrestrictedTraverse(self.object_path)
    
  def getObjectList(self, activity_tool):
107 108 109 110 111 112 113
    try:
      expand_method_id = self.activity_kw['expand_method_id']
      obj = self.getObject(activity_tool)
      # FIXME: how to pass parameters?
      object_list = getattr(obj, expand_method_id)()
    except KeyError:
      object_list = [self.getObject(activity_tool)]
114
      
115
    return object_list
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137
      
  def hasExpandMethod(self):
    return self.activity_kw.has_key('expand_method_id')
    
  def changeUser(self, user_name, activity_tool):
    uf = activity_tool.getPortalObject().acl_users
    user = uf.getUserById(user_name)
    if user is not None:
      user = user.__of__(uf)
      newSecurityManager(None, user)
    return user

  def activateResult(self, activity_tool, result, object):
    if self.active_process is not None:
      active_process = activity_tool.unrestrictedTraverse(self.active_process)
      if isinstance(result,Error):
        result.edit(object_path=object)
        result.edit(method_id=self.method_id)
        active_process.activateResult(result) # XXX Allow other method_id in future
      else:
        active_process.activateResult(Error(object_path=object,method_id=self.method_id,result=result)) # XXX Allow other method_id in future
  
Jean-Paul Smets's avatar
Jean-Paul Smets committed
138
  def __call__(self, activity_tool):
139
    try:
140 141
#       LOG('WARNING ActivityTool', 0,
#            'Trying to call method %s on object %s' % (self.method_id, self.object_path))
142
      obj = self.getObject(activity_tool)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
143
      # Change user if required (TO BE DONE)
144 145
      # We will change the user only in order to execute this method
      current_user = str(_getAuthenticatedUser(self))
146
      user = self.changeUser(self.user_name, activity_tool)
147 148 149 150 151 152 153
      try:
        result = getattr(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)
      self.activateResult(activity_tool, result, obj)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
154
      self.is_executed = 1
155
    except:
Jean-Paul Smets's avatar
Jean-Paul Smets committed
156
      self.is_executed = 0
157
      self.exc_type = sys.exc_info()[0]
158 159
      LOG('WARNING ActivityTool', 0,
          'Could not call method %s on object %s' % (self.method_id, self.object_path), error=sys.exc_info())
Jean-Paul Smets's avatar
Jean-Paul Smets committed
160 161 162 163

  def validate(self, activity, activity_tool):
    return activity.validate(activity_tool, self, **self.activity_kw)

164
  def notifyUser(self, activity_tool, message="Failed Processing Activity"):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
165 166 167
    #LOG('notifyUser begin', 0, str(self.user_name))
    user_email = activity_tool.portal_membership.getMemberById(self.user_name).getProperty('email')
    if user_email in ('', None):
168
      user_email = getattr(activity_tool, 'email_to_address', activity_tool.email_from_address)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
169 170
    #LOG('notifyUser user_email', 0, str(user_email))
    mail_text = """From: %s
171 172 173 174 175 176 177 178 179
To: %s
Subject: %s

%s

Document: %s
Method: %s
    """ % (activity_tool.email_from_address, user_email,
           message, message, '/'.join(self.object_path), self.method_id)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
180 181 182
    #LOG('notifyUser mail_text', 0, str(mail_text))
    activity_tool.MailHost.send( mail_text )
    #LOG('notifyUser send', 0, '')
183

184 185 186 187 188 189 190 191 192 193 194 195 196 197 198
  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
199 200
class Method:

201
  def __init__(self, passive_self, activity, active_process, kw, method_id):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
202 203
    self.__passive_self = passive_self
    self.__activity = activity
204
    self.__active_process = active_process
Jean-Paul Smets's avatar
Jean-Paul Smets committed
205 206 207 208
    self.__kw = kw
    self.__method_id = method_id

  def __call__(self, *args, **kw):
209
    m = Message(self.__passive_self, self.__active_process, self.__kw, self.__method_id, args, kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
210 211
    activity_dict[self.__activity].queueMessage(self.__passive_self.portal_activities, m)

212 213
allow_class(Method)

Jean-Paul Smets's avatar
Jean-Paul Smets committed
214 215
class ActiveWrapper:

216
  def __init__(self, passive_self, activity, active_process, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
217 218
    self.__dict__['__passive_self'] = passive_self
    self.__dict__['__activity'] = activity
219
    self.__dict__['__active_process'] = active_process
Jean-Paul Smets's avatar
Jean-Paul Smets committed
220 221 222 223
    self.__dict__['__kw'] = kw

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

227
class ActivityTool (Folder, UniqueObject):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
228
    """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
229 230 231 232 233 234 235 236 237 238 239 240
    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
241 242 243
    """
    id = 'portal_activities'
    meta_type = 'CMF Activity Tool'
244
    portal_type = 'Activity Tool'
245
    allowed_types = ( 'CMF Active Process', )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
246 247
    security = ClassSecurityInfo()

248 249 250
    _distributingNode = ''
    _nodes = ()

251 252
    manage_options = tuple(
                     [ { 'label' : 'Overview', 'action' : 'manage_overview' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
253
                     , { 'label' : 'Activities', 'action' : 'manageActivities' }
254
                     , { 'label' : 'LoadBalancing', 'action' : 'manageLoadBalancing'}
255
                     , { 'label' : 'Advanced', 'action' : 'manageActivitiesAdvanced' }
Jean-Paul Smets's avatar
Jean-Paul Smets committed
256
                     ,
257
                     ] + list(Folder.manage_options))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
258 259 260 261

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

262 263 264
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageActivitiesAdvanced' )
    manageActivitiesAdvanced = DTMLFile( 'dtml/manageActivitiesAdvanced', globals() )

265 266
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manage_overview' )
    manage_overview = DTMLFile( 'dtml/explainActivityTool', globals() )
267 268 269 270 271 272
    
    security.declareProtected( CMFCorePermissions.ManagePortal , 'manageLoadBalancing' )
    manageLoadBalancing = DTMLFile( 'dtml/manageLoadBalancing', globals() )
    
    distributingNode = ''
    _nodes = ()
273 274 275

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

277 278 279 280 281 282 283 284 285 286
    # 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
287 288
    def initialize(self):
      global is_initialized
Sebastien Robin's avatar
Sebastien Robin committed
289
      from Activity import RAMQueue, RAMDict, SQLQueue, SQLDict
Jean-Paul Smets's avatar
Jean-Paul Smets committed
290 291 292 293
      # Initialize each queue
      for activity in activity_list:
        activity.initialize(self)
      is_initialized = 1
294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
      
    security.declareProtected(Permissions.manage_properties, 'isSubscribed')
    def isSubscribed(self):
        """"
        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
310

311 312 313 314
    security.declareProtected(Permissions.manage_properties, 'subscribe')
    def subscribe(self):
        """ subscribe to the global Timer Service """
        service = getTimerService(self)
315
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
316
        if not service:
317
            LOG('ActivityTool', INFO, 'TimerService not available')
318 319 320 321 322
            url += urllib.quote('TimerService not available')
        else:
            service.subscribe(self)
            url += urllib.quote("Subscribed to Timer Service")
        return self.REQUEST.RESPONSE.redirect(url)
323 324 325 326 327

    security.declareProtected(Permissions.manage_properties, 'unsubscribe')
    def unsubscribe(self):
        """ unsubscribe from the global Timer Service """
        service = getTimerService(self)
328
        url = '%s/manageLoadBalancing?manage_tabs_message=' %self.absolute_url()
329
        if not service:
330
            LOG('ActivityTool', INFO, 'TimerService not available')
331 332 333 334 335
            url += urllib.quote('TimerService not available')
        else:
            service.unsubscribe(self)
            url += urllib.quote("Unsubscribed from Timer Service")
        return self.REQUEST.RESPONSE.redirect(url)
336 337 338

    def manage_beforeDelete(self, item, container):
        self.unsubscribe()
339 340
        Folder.inheritedAttribute('manage_beforeDelete')(self, item, container)
    
341 342
    def manage_afterAdd(self, item, container):
        self.subscribe()
343 344
        Folder.inheritedAttribute('manage_afterAdd')(self, item, container)
       
345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412
    def getCurrentNode(self):
        """ Return current node in form ip:port """
        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)
        return currentNode
        
    security.declarePublic('getDistributingNode')
    def getDistributingNode(self):
        """ Return the distributingNode """
        return self.distributingNode

    security.declarePublic('getNodeList')
    def getNodes(self):
        """ Return all nodes """
        return self._nodes

    security.declarePublic('manage_setDistributingNode')
    def manage_setDistributingNode(self, distributingNode, REQUEST=None):
        """ set the distributing node """
        self.distributingNode = distributingNode
        if REQUEST is not None:
            REQUEST.RESPONSE.redirect(
                REQUEST.URL1 +
                '/manageLoadBalancing?manage_tabs_message=' +
                urllib.quote("Distributing Node successfully changed."))
    
    security.declarePublic('manage_addNode')
    def manage_addNode(self, node, REQUEST=None):
        """ add a node """
        if node in self._nodes:
            if REQUEST is not None:
                REQUEST.RESPONSE.redirect(
                    REQUEST.URL1 +
                    '/manageLoadBalancing?manage_tabs_message=' +
                    urllib.quote("Node exists already."))
            return
            
        self._nodes = self._nodes + (node,)
        
        if REQUEST is not None:
            REQUEST.RESPONSE.redirect(
                REQUEST.URL1 +
                '/manageLoadBalancing?manage_tabs_message=' +
                urllib.quote("Node successfully added."))
                        
    security.declarePublic('manage_delNode')
    def manage_delNode(self, deleteNodes, REQUEST=None):
        """ delete nodes """
        nodeList = list(self._nodes)
        for node in deleteNodes:
            if node in self._nodes:
                nodeList.remove(node)
        self._nodes = tuple(nodeList)
        if REQUEST is not None:
            REQUEST.RESPONSE.redirect(
                REQUEST.URL1 +
                '/manageLoadBalancing?manage_tabs_message=' +
                urllib.quote("Node(s) successfully deleted."))
        
413
    def process_timer(self, tick, interval, prev="", next=""):
414 415 416 417 418 419
        """ 
        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.
        """
420 421 422 423
        # Prevent TimerService from starting multiple threads in parallel
        acquired = timerservice_lock.acquire(0)
        if not acquired:
          return
424

425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457
        try:
          # get owner of portal_catalog, so normally we should be able to
          # have the permission to invoke all activities
          user = self.portal_catalog.getOwner()
          newSecurityManager(self.REQUEST, user)
          
          currentNode = self.getCurrentNode()
          
          # only distribute when we are the distributingNode or if it's empty
          if (self.distributingNode == self.getCurrentNode()):
              self.distribute(len(self._nodes))
              #LOG('CMFActivity:', INFO, 'self.distribute(node_count=%s)' %len(self._nodes))

          elif not self.distributingNode:
              self.distribute(1)
              #LOG('CMFActivity:', INFO, 'distributingNodes empty! Calling distribute(1)')
          
          # 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 self._nodes:
              self.tic(list(self._nodes).index(currentNode)+1)
              #LOG('CMFActivity:', INFO, 'self.tic(processing_node=%s)' %str(list(self._nodes).index(currentNode)+1))
              
          elif len(self._nodes) == 0:
              self.tic(1)
              #LOG('CMFActivity:', INFO, 'Node List is empty! Calling tic(1)')

        except:
          timerservice_lock.release()
          raise
        else:
          timerservice_lock.release()
458

Jean-Paul Smets's avatar
Jean-Paul Smets committed
459 460 461 462 463 464
    security.declarePublic('distribute')
    def distribute(self, node_count=1):
      """
        Distribute load
      """
      # Initialize if needed
465
      global is_initialized
Jean-Paul Smets's avatar
Jean-Paul Smets committed
466 467 468 469
      if not is_initialized: self.initialize()

      # Call distribute on each queue
      for activity in activity_list:
470
        try:
Jean-Paul Smets's avatar
Jean-Paul Smets committed
471
          activity.distribute(self, node_count)
472 473
        except ConflictError:
          raise
474
        except:
Yoshinori Okuji's avatar
Yoshinori Okuji committed
475
          LOG('CMFActivity:', 100, 'Core call to distribute failed for activity %s' % activity, error=sys.exc_info())
Jean-Paul Smets's avatar
Jean-Paul Smets committed
476

Jean-Paul Smets's avatar
Jean-Paul Smets committed
477
    security.declarePublic('tic')
Jean-Paul Smets's avatar
Jean-Paul Smets committed
478
    def tic(self, processing_node=1, force=0):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
479 480
      """
        Starts again an activity
Jean-Paul Smets's avatar
Jean-Paul Smets committed
481
        processing_node starts from 1 (there is not node 0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
482
      """
483
      global active_threads, is_initialized, first_run
Jean-Paul Smets's avatar
Jean-Paul Smets committed
484 485

      # return if the number of threads is too high
486
      # else, increase the number of active_threads and continue
487 488
      tic_lock.acquire()
      too_many_threads = (active_threads >= max_active_threads)
489
      if not too_many_threads or force:
490
        active_threads += 1
491 492 493
      else:
        tic_lock.release()
        raise RuntimeError, 'Too many threads'
494
      tic_lock.release()
495

Jean-Paul Smets's avatar
Jean-Paul Smets committed
496 497 498
      # Initialize if needed
      if not is_initialized: self.initialize()

499 500 501 502 503 504 505
      # If this is the first tic after zope is started, reset the processing
      # flag for activities of this node
      if first_run:
        self.SQLDict_clearProcessingFlag(processing_node=processing_node)
        self.SQLQueue_clearProcessingFlag(processing_node=processing_node)
        first_run = 0

506 507
      try:
        # Wakeup each queue
Jean-Paul Smets's avatar
Jean-Paul Smets committed
508
        for activity in activity_list:
509
          try:
510 511 512
            activity.wakeup(self, processing_node)
          except ConflictError:
            raise
513
          except:
514 515 516 517 518 519 520 521 522 523
            LOG('CMFActivity:', 100, 'Core call to wakeup failed for activity %s' % activity)
  
        # Process messages on each queue in round robin
        has_awake_activity = 1
        while has_awake_activity:
          has_awake_activity = 0
          for activity in activity_list:
            try:
              activity.tic(self, processing_node) # Transaction processing is the responsability of the activity
              has_awake_activity = has_awake_activity or activity.isAwake(self, processing_node)
524
              #LOG('ActivityTool tic', 0, 'has_awake_activity = %r, activity = %r, activity.isAwake(self, processing_node) = %r' % (has_awake_activity, activity, activity.isAwake(self, processing_node)))
525 526 527 528 529 530 531 532 533
            except ConflictError:
              raise
            except:
              LOG('CMFActivity:', 100, 'Core call to tic or isAwake failed for activity %s' % activity, error=sys.exc_info())
      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
534

535
    def hasActivity(self, *args, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
536
      # Check in each queue if the object has deferred tasks
537 538 539 540 541
      # if not argument is provided, then check on self
      if len(args) > 0:
        object = args[0]
      else:
        object = self
Jean-Paul Smets's avatar
Jean-Paul Smets committed
542 543 544 545 546
      for activity in activity_list:
        if activity.hasActivity(self, object, **kw):
          return 1
      return 0

547
    def activate(self, object, activity, active_process, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
548 549
      global is_initialized
      if not is_initialized: self.initialize()
550
      if not hasattr(self, '_v_activity_buffer'): self._v_activity_buffer = ActivityBuffer()
551
      return ActiveWrapper(object, activity, active_process, **kw)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
552

553 554
    def deferredQueueMessage(self, activity, message):
      self._v_activity_buffer.deferredQueueMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
555

556
    def deferredDeleteMessage(self, activity, message):
557
      if not hasattr(self, '_v_activity_buffer'): self._v_activity_buffer = ActivityBuffer()
558
      self._v_activity_buffer.deferredDeleteMessage(self, activity, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
559

Jean-Paul Smets's avatar
Jean-Paul Smets committed
560
    def getRegisteredMessageList(self, activity):
561 562 563 564
      activity_buffer = getattr(self, '_v_activity_buffer', None)
      #if getattr(self, '_v_activity_buffer', None):
      if activity_buffer is not None:
        activity_buffer._register() # This is required if flush flush is called outside activate
565 566 567
        return activity.getRegisteredMessageList(self._v_activity_buffer, self)
      else:
        return []
Yoshinori Okuji's avatar
Yoshinori Okuji committed
568

Jean-Paul Smets's avatar
Jean-Paul Smets committed
569
    def unregisterMessage(self, activity, message):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
570
      self._v_activity_buffer._register() # Required if called by flush, outside activate
Jean-Paul Smets's avatar
Jean-Paul Smets committed
571
      return activity.unregisterMessage(self._v_activity_buffer, self, message)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
572

Jean-Paul Smets's avatar
Jean-Paul Smets committed
573 574 575
    def flush(self, object, invoke=0, **kw):
      global is_initialized
      if not is_initialized: self.initialize()
576
      if not hasattr(self, '_v_activity_buffer'): self._v_activity_buffer = ActivityBuffer()
Yoshinori Okuji's avatar
Yoshinori Okuji committed
577
      if type(object) is TupleType:
578 579 580
        object_path = object
      else:
        object_path = object.getPhysicalPath()
Jean-Paul Smets's avatar
Jean-Paul Smets committed
581
      for activity in activity_list:
582
#         LOG('CMFActivity: ', 0, 'flushing activity %s' % activity.__class__.__name__)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
583 584
        activity.flush(self, object_path, invoke=invoke, **kw)

585 586 587 588
    def start(self, **kw):
      global is_initialized
      if not is_initialized: self.initialize()
      for activity in activity_list:
589
#         LOG('CMFActivity: ', 0, 'starting activity %s' % activity.__class__.__name__)
590 591 592 593 594 595
        activity.start(self, **kw)

    def stop(self, **kw):
      global is_initialized
      if not is_initialized: self.initialize()
      for activity in activity_list:
596
#         LOG('CMFActivity: ', 0, 'starting activity %s' % activity.__class__.__name__)
597 598
        activity.stop(self, **kw)

Jean-Paul Smets's avatar
Jean-Paul Smets committed
599 600
    def invoke(self, message):
      message(self)
601 602 603 604 605 606 607 608 609 610 611 612
      
    def invokeGroup(self, method_id, message_list):
      # Invoke a group method.
      object_list = []
      expanded_object_list = []
      new_message_list = []
      path_dict = {}
      # Filter the list of messages. If an object is not available, ignore such a message.
      # In addition, expand an object if necessary, and make sure that no duplication happens.
      for m in message_list:
        try:
          obj = m.getObject(self)
613
          i = len(new_message_list) # This is an index of this message in new_message_list.
614
          if m.hasExpandMethod():
615 616
            for subobj in m.getObjectList(self):
              path = subobj.getPath()
617
              if path not in path_dict:
618 619
                path_dict[path] = i
                expanded_object_list.append(subobj)
620 621 622
          else:
            path = obj.getPath()
            if path not in path_dict:
623
              path_dict[path] = i
624
              expanded_object_list.append(obj)
625
          object_list.append(obj)
626 627 628
          new_message_list.append(m)
        except:
          m.is_executed = 0
629
          m.exc_type = sys.exc_info()[0]
630 631 632 633 634 635 636 637
          LOG('WARNING ActivityTool', 0,
              'Could not call method %s on object %s' % (m.method_id, m.object_path), error=sys.exc_info())
              
      if len(expanded_object_list) > 0:
        try:
          method = self.unrestrictedTraverse(method_id)
          # FIXME: how to pass parameters?
          # FIXME: how to apply security here?
638 639
          # 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.
640 641
          result = method(expanded_object_list)
        except:
642
          # In this case, the group method completely failed.
643 644
          for m in new_message_list:
            m.is_executed = 0
645
            m.exc_type = sys.exc_info()[0]
646 647 648
          LOG('WARNING ActivityTool', 0,
              'Could not call method %s on objects %s' % (method_id, expanded_object_list), error=sys.exc_info())
        else:
649 650 651 652 653 654 655 656
          # 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).
657 658 659
          for i in xrange(len(object_list)):
            object = object_list[i]
            m = new_message_list[i]
660
            if i in failed_message_dict:
661
              m.is_executed = 0
662 663 664 665 666 667 668 669
              LOG('WARNING ActivityTool', 0, 
                  'the method %s partially failed on object %s' % (m.method_id, m.object_path,))
            else:
              try:
                m.activateResult(self, result, object)
                m.is_executed = 1
              except:
                m.is_executed = 0
670
                m.exc_type = sys.exc_info()[0]
671 672
                LOG('WARNING ActivityTool', 0,
                    'Could not call method %s on object %s' % (m.method_id, m.object_path), error=sys.exc_info())
673
            
674 675
    def newMessage(self, activity, path, active_process, activity_kw, method_id, *args, **kw):
      # Some Security Cheking should be made here XXX
Jean-Paul Smets's avatar
Jean-Paul Smets committed
676 677
      global is_initialized
      if not is_initialized: self.initialize()
678
      if not hasattr(self, '_v_activity_buffer'): self._v_activity_buffer = ActivityBuffer()
679
      activity_dict[activity].queueMessage(self, Message(path, active_process, activity_kw, method_id, args, kw))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
680

681
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageInvoke' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
682 683 684 685 686 687
    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('/'))
688
      self.flush(object_path,method_id=method_id,invoke=1)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
689 690 691
      if REQUEST is not None:
        return REQUEST.RESPONSE.redirect('%s/%s' % (self.absolute_url(), 'manageActivities'))

692
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageCancel' )
Jean-Paul Smets's avatar
Jean-Paul Smets committed
693 694 695 696 697 698
    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('/'))
699
      self.flush(object_path,method_id=method_id,invoke=0)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
700 701 702
      if REQUEST is not None:
        return REQUEST.RESPONSE.redirect('%s/%s' % (self.absolute_url(), 'manageActivities'))

703
    security.declareProtected( CMFCorePermissions.ManagePortal, 'manageClearActivities' )
704
    def manageClearActivities(self, keep=1, REQUEST=None):
705 706 707 708 709
      """
        Clear all activities and recreate tables.
      """
      folder = getToolByName(self, 'portal_skins').activity

710 711
      # Obtain all pending messages.
      message_list = []
712 713 714 715 716 717 718 719 720
      if keep:
        for activity in activity_list:
          if hasattr(activity, 'dumpMessageList'):
            try:
              message_list.extend(activity.dumpMessageList(self))
            except ConflictError:
              raise
            except:
              LOG('ActivityTool', WARNING, 'could not dump messages from %s' % (activity,), error=sys.exc_info())
721
            
722 723 724 725 726 727 728
      if hasattr(folder, 'SQLDict_createMessageTable'):
        try:
          folder.SQLDict_dropMessageTable()
        except ConflictError:
          raise
        except:
          LOG('CMFActivities', 
729 730
              WARNING, 
              'could not drop the message table',
731 732 733 734 735 736 737 738 739 740
              error=sys.exc_info())
        folder.SQLDict_createMessageTable()

      if hasattr(folder, 'SQLQueue_createMessageTable'):
        try:
          folder.SQLQueue_dropMessageTable()
        except ConflictError:
          raise
        except:
          LOG('CMFActivities', 
741 742
              WARNING, 
              'could not drop the message queue table',
743 744 745
              error=sys.exc_info())
        folder.SQLQueue_createMessageTable()

746 747 748 749 750 751 752 753 754 755 756
      # Reactivate the messages.
      for m in message_list:
        try:
          m.reactivate(self)
        except ConflictError:
          raise
        except:
          LOG('ActivityTool', WARNING,
              'could not reactivate the message %r, %r' % (m.object_path, m.method_id),
              error=sys.exc_info())

757 758 759
      if REQUEST is not None:
        return REQUEST.RESPONSE.redirect('%s/%s' % (self.absolute_url(), 'manageActivitiesAdvanced?manage_tabs_message=Activities%20Cleared')) 

Jean-Paul Smets's avatar
Jean-Paul Smets committed
760 761 762 763 764
    security.declarePublic('getMessageList')
    def getMessageList(self):
      """
        List messages waiting in queues
      """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
765 766 767
      # Initialize if needed
      if not is_initialized: self.initialize()

Jean-Paul Smets's avatar
Jean-Paul Smets committed
768 769
      message_list = []
      for activity in activity_list:
Sebastien Robin's avatar
Sebastien Robin committed
770 771 772 773
        try:
          message_list += activity.getMessageList(self)
        except AttributeError:
          LOG('getMessageList, could not get message from Activity:',0,activity)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
774 775
      return message_list

776
    security.declareProtected( CMFCorePermissions.ManagePortal , 'newActiveProcess' )
777
    def newActiveProcess(self, **kw):
778 779 780
      from ActiveProcess import addActiveProcess
      new_id = str(self.generateNewId())
      addActiveProcess(self, new_id)
781 782 783
      active_process = self._getOb(new_id)
      active_process.edit(**kw)
      return active_process
784 785 786 787

    def reindexObject(self):
      self.immediateReindexObject()

788 789 790 791 792 793 794
    # Active synchronisation methods
    def validateOrder(self, message, validator_id, validation_value):
      global is_initialized
      if not is_initialized: self.initialize()
      for activity in activity_list:
        method_id = "_validate_%s" % validator_id
        if hasattr(activity, method_id):
795
#           LOG('CMFActivity: ', 0, 'validateOrder calling method_id %s' % method_id)
796 797 798
          if getattr(activity,method_id)(self, message, validation_value):
            return 1
      return 0
799

Yoshinori Okuji's avatar
Yoshinori Okuji committed
800 801
    # Required for tests (time shift)
    def timeShift(self, delay):
802 803 804 805
      global is_initialized
      if not is_initialized: self.initialize()
      for activity in activity_list:
        activity.timeShift(self, delay)
Yoshinori Okuji's avatar
Yoshinori Okuji committed
806

807
InitializeClass(ActivityTool)