opensource/SpiffWorkflow
1
0
Fork 0
mirror of https://github.com/sartography/SpiffWorkflow.git synced 2025-08-02 00:59:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

add update method on event definition

Browse Source
This commit is contained in:
Elizabeth Esswein
2024-03-27 11:06:10 -04:00
parent 0010df0ac6
commit 35a015de8d
3 changed files with 14 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
SpiffWorkflow/bpmn/specs/event_definitions/base.py
View File

@ -29,8 +29,12 @@ class EventDefinition(object):
event = BpmnEvent(self)
my_task.workflow.top_workflow.catch(event)
def update_task(self, my_task):
"""This method allows events to implement update behavior for the task"""
pass
def update_task_data(self, my_task):
"""This method allows events with payloads mrege them into the task"""
"""This method allows events with payloads to merge them into the task"""
pass
def reset(self, my_task):

8
SpiffWorkflow/bpmn/specs/event_definitions/timer.py
View File

@ -3,6 +3,7 @@ from datetime import datetime, timedelta, timezone
from calendar import monthrange
from time import timezone as tzoffset, altzone as dstoffset, struct_time, localtime
from SpiffWorkflow.util.task import TaskState
from SpiffWorkflow.bpmn.util import PendingBpmnEvent
from .base import EventDefinition
@ -200,6 +201,13 @@ class CycleTimerEventDefinition(TimerEventDefinition):
my_task._set_internal_data(event_value=event_value)
return ready
def update_task(self, my_task):
if self.cycle_complete(my_task):
for output in my_task.task_spec.outputs:
child = my_task._add_child(output, TaskState.FUTURE)
child.task_spec._predict(child, mask=TaskState.NOT_FINISHED_MASK)
child.task_spec._update(child)
def details(self, my_task):
event_value = my_task._get_internal_data('event_value')
if event_value is not None and event_value['cycles'] != 0:

8
SpiffWorkflow/bpmn/specs/mixins/events/event_types.py
View File

@ -60,13 +60,7 @@ class CatchingEvent(TaskSpec):
elif my_task.state != TaskState.WAITING:
my_task._set_state(TaskState.WAITING)
if isinstance(self.event_definition, CycleTimerEventDefinition):
if self.event_definition.cycle_complete(my_task):
for output in self.outputs:
child = my_task._add_child(output, TaskState.FUTURE)
child.task_spec._predict(child, mask=TaskState.NOT_FINISHED_MASK)
child.task_spec._update(child)
self.event_definition.update_task(my_task)
def _run_hook(self, my_task):