Feature/event subprocess

Makefile

@@ -50,10 +50,6 @@ tests-ind:
 tests-timing:
 	@make tests-ind 2>&1 | ./scripts/test_times.py
 
-.PHONY : waiting-task-stress
-waiting-task-stress:
-	$(RUN) python -m unittest -v tests.SpiffWorkflow.bpmn.WaitingTaskStressBenchmark
-
 
 wheel: clean
 	$(RUN) python -m build --sdist --wheel --outdir dist/

SpiffWorkflow/bpmn/parser/BpmnParser.py

@@ -45,11 +45,8 @@
     BoundaryEvent,
     EventBasedGateway
 )
-from SpiffWorkflow.bpmn.specs.event_definitions.simple import NoneEventDefinition
-from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition
 from SpiffWorkflow.bpmn.specs.event_definitions.message import CorrelationProperty
 from SpiffWorkflow.bpmn.specs.mixins.subworkflow_task import SubWorkflowTask as SubWorkflowTaskMixin
-from SpiffWorkflow.bpmn.specs.mixins.events.start_event import StartEvent as StartEventMixin
 
 from .ValidationException import ValidationException
 from .ProcessParser import ProcessParser
@@ -448,23 +445,13 @@ def get_collaboration(self, name):
         spec = BpmnProcessSpec(name)
         subprocesses = {}
         participant_type = self._get_parser_class(full_tag('callActivity'))[1]
-        start_type = self._get_parser_class(full_tag('startEvent'))[1]
-        end_type = self._get_parser_class(full_tag('endEvent'))[1]
-        start = start_type(spec, 'Start Collaboration', NoneEventDefinition())
-        spec.start.connect(start)
-        end = end_type(spec, 'End Collaboration', NoneEventDefinition())
-        end.connect(spec.end)
         for process in self.collaborations[name]:
             process_parser = self.get_process_parser(process)
             if process_parser and process_parser.process_executable:
                 sp_spec = self.get_spec(process)
                 subprocesses[process] = sp_spec
                 subprocesses.update(self.get_subprocess_specs(process))
-                if len([s for s in sp_spec.task_specs.values() if
-                        isinstance(s, StartEventMixin) and
-                        isinstance(s.event_definition, (NoneEventDefinition, TimerEventDefinition))
-                ]):
-                    participant = participant_type(spec, process, process)
-                    start.connect(participant)
-                    participant.connect(end)
+                participant = participant_type(spec, process, process)
+                spec.start.connect_or_add_trigger(participant, sp_spec)
+
         return spec, subprocesses

SpiffWorkflow/bpmn/parser/ProcessParser.py

@@ -191,6 +191,11 @@ def _parse(self):
             self.spec.start.outputs = [split_task]
             split_task.inputs = [self.spec.start]
 
+        for node in self.xpath("./bpmn:subProcess[@triggeredByEvent='true']"):
+            task_spec = self.parse_node(node)
+            sp_spec = self.parser.process_parsers[task_spec.spec].get_spec()
+            self.spec.start.connect_or_add_trigger(task_spec, sp_spec)
+
     def parse_data_object(self, obj):
         return self.create_data_spec(obj, DataObject)
 

SpiffWorkflow/bpmn/parser/TaskParser.py

@@ -23,7 +23,8 @@
 from SpiffWorkflow.bpmn.specs.defaults import (
     StandardLoopTask,
     SequentialMultiInstanceTask,
-    ParallelMultiInstanceTask
+    ParallelMultiInstanceTask,
+    EventSubprocess,
 )
 from SpiffWorkflow.bpmn.specs.control import BoundaryEventSplit, BoundaryEventJoin
 from SpiffWorkflow.bpmn.specs.event_definitions.simple import CancelEventDefinition
@@ -47,6 +48,8 @@ class TaskParser(NodeParser):
     STANDARD_LOOP_CLASS = StandardLoopTask
     PARALLEL_MI_CLASS = ParallelMultiInstanceTask
     SEQUENTIAL_MI_CLASS = SequentialMultiInstanceTask
+    # I have to add another attribute here.  This parser is so stupid.
+    EVENT_SUBPROCESS_CLASS = EventSubprocess
 
     def __init__(self, process_parser, spec_class, node, nsmap=None, lane=None):
         """

SpiffWorkflow/bpmn/parser/task_parsers.py

@@ -94,7 +94,11 @@ class SubWorkflowParser(TaskParser):
 
     def create_task(self):
         subworkflow_spec = SubprocessParser.get_subprocess_spec(self)
-        return self.spec_class(self.spec, self.bpmn_id, subworkflow_spec=subworkflow_spec, **self.bpmn_attributes)
+        if self.attribute('triggeredByEvent'):
+            spec_class = self.EVENT_SUBPROCESS_CLASS
+        else:
+            spec_class = self.spec_class
+        return spec_class(self.spec, self.bpmn_id, subworkflow_spec=subworkflow_spec, **self.bpmn_attributes)
 
 
 class CallActivityParser(TaskParser):

SpiffWorkflow/bpmn/serializer/config.py

@@ -37,6 +37,7 @@
     SubWorkflowTask,
     CallActivity,
     TransactionSubprocess,
+    EventSubprocess,
     StartEvent,
     EndEvent,
     IntermediateCatchEvent,
@@ -96,6 +97,7 @@
     EventConverter,
     BoundaryEventConverter,
     IOSpecificationConverter,
+    BpmnStartTaskConverter,
 )
 from .default.event_definition import (
     TimerConditionalEventDefinitionConverter,
@@ -115,7 +117,7 @@
     BpmnIoSpecification: IOSpecificationConverter,
     BpmnProcessSpec: BpmnProcessSpecConverter,
     SimpleBpmnTask: BpmnTaskSpecConverter,
-    BpmnStartTask: BpmnTaskSpecConverter,
+    BpmnStartTask: BpmnStartTaskConverter,
     _EndJoin: BpmnTaskSpecConverter,
     NoneTask: BpmnTaskSpecConverter,
     ManualTask: BpmnTaskSpecConverter,
@@ -127,6 +129,7 @@
     SubWorkflowTask: SubWorkflowConverter,
     CallActivity: SubWorkflowConverter,
     TransactionSubprocess: SubWorkflowConverter,
+    EventSubprocess: SubWorkflowConverter,
     BoundaryEventSplit: BpmnTaskSpecConverter,
     BoundaryEventJoin: EventJoinConverter,
     ExclusiveGateway: ExclusiveGatewayConverter,

SpiffWorkflow/bpmn/serializer/default/process_spec.py

@@ -36,6 +36,7 @@ def to_dict(self, spec):
             'io_specification': self.registry.convert(spec.io_specification),
             'data_objects': {name: self.registry.convert(obj) for name, obj in spec.data_objects.items()},
             'correlation_keys': spec.correlation_keys,
+            'bpmn_start_events': [ts.name for ts in spec.bpmn_start_events],
         }
         for name, task_spec in spec.task_specs.items():
             task_dict = self.registry.convert(task_spec)
@@ -82,4 +83,6 @@ def from_dict(self, dct):
             child_spec = spec.task_specs.get(task_spec.task_spec)
             child_spec.completed_event.connect(task_spec.merge_child)
 
+        spec.bpmn_start_events = [spec.task_specs.get(name) for name in dct.get('bpmn_start_events', [])]
+
         return spec

SpiffWorkflow/bpmn/serializer/default/task_spec.py

@@ -201,6 +201,18 @@ def to_dict(self, spec):
     def from_dict(self, dct):
         return self.task_spec_from_dict(dct)
 
+class BpmnStartTaskConverter(BpmnTaskSpecConverter):
+
+    def to_dict(self, spec):
+        dct = super().to_dict(spec)
+        dct['trigger_specs'] = self.registry.convert(spec.trigger_specs)
+        return dct
+
+    def from_dict(self, dct):
+        trigger_specs = dct.pop('trigger_specs', [])
+        spec = super().from_dict(dct)
+        spec.trigger_specs = trigger_specs
+        return spec
 
 class EventConverter(BpmnTaskSpecConverter):
     """The default converter for BPMN events"""

SpiffWorkflow/bpmn/serializer/default/workflow.py

@@ -19,7 +19,9 @@
 
 from uuid import UUID
 
+from SpiffWorkflow.task import TaskState
 from SpiffWorkflow.bpmn.specs.mixins.subworkflow_task import SubWorkflowTask
+from SpiffWorkflow.bpmn.specs.mixins.events.event_types import CatchingEvent
 from SpiffWorkflow.util.deep_merge import DeepMerge
 
 from ..helpers.bpmn_converter import BpmnConverter
@@ -62,6 +64,8 @@ def from_dict(self, dct, workflow):
         task.last_state_change = dct['last_state_change']
         task.triggered = dct['triggered']
         task.internal_data = self.registry.restore(dct['internal_data'])
+        if isinstance(task_spec, CatchingEvent) and task.has_state(TaskState.WAITING):
+            task.workflow.top_workflow.event_manager.add_task(task)
 
         delta = dct.get('delta')
         if delta and task.parent is not None:
@@ -101,6 +105,8 @@ def from_dict(self, dct, workflow):
         task.triggered = dct['triggered']
         task.internal_data = self.registry.restore(dct['internal_data'])
         task.data = self.registry.restore(dct['data'])
+        if isinstance(task_spec, CatchingEvent) and task.has_state(TaskState.DEFINITE_MASK):
+            task.workflow.top_workflow.event_manager.add_task(task)
         return task
 
 class BpmnEventConverter(BpmnConverter):
@@ -201,7 +207,6 @@ def from_dict(self, dct):
         # Handle the remaining top workflow attributes
         self.subprocesses_from_dict(dct['subprocesses'], workflow)
         workflow.bpmn_events = self.registry.restore(dct.pop('bpmn_events', []))
-        workflow._rebuild_waiting_task_index()
 
         return workflow
 

SpiffWorkflow/bpmn/specs/bpmn_process_spec.py

@@ -19,6 +19,7 @@
 
 from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
 from SpiffWorkflow.bpmn.specs.control import _EndJoin, BpmnStartTask, SimpleBpmnTask
+from SpiffWorkflow.bpmn.specs.mixins.events.start_event import StartEvent
 
 
 class BpmnProcessSpec(WorkflowSpec):
@@ -45,3 +46,10 @@ def __init__(self, name=None, description=None, filename=None, svg=None):
         self.data_objects = {}
         self.data_stores = {}
         self.correlation_keys = {}
+        self.bpmn_start_events = []
+
+    def _add_notify(self, task_spec):
+        super()._add_notify(task_spec)
+        if isinstance(task_spec, (StartEvent, )):
+            self.bpmn_start_events.append(task_spec)
+

SpiffWorkflow/bpmn/specs/control.py

@@ -27,9 +27,35 @@
 from SpiffWorkflow.bpmn.specs.mixins.events.intermediate_event import BoundaryEvent
 from SpiffWorkflow.bpmn.specs.mixins.events.start_event import StartEvent
 
+from SpiffWorkflow.bpmn.specs.event_definitions.simple import NoneEventDefinition
+from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition
+
 
 class BpmnStartTask(BpmnTaskSpec, StartTask):
-    pass
+
+    def __init__(self, wf_spec, name, **kwargs):
+        super().__init__(wf_spec, name, **kwargs)
+        self.trigger_specs = []
+
+    def connect_or_add_trigger(self, task_spec, sp_spec):
+        for ts in sp_spec.bpmn_start_events:
+            if isinstance(ts.event_definition, (NoneEventDefinition, TimerEventDefinition)):
+                self.connect(task_spec)
+                task_spec.connect(self._wf_spec.end)
+            else:
+                self._wf_spec.task_specs[sp_spec.name] = task_spec
+                self.trigger_specs.append(sp_spec.name)
+
+    def trigger_wf(self, my_task, spec_name):
+        spec = self._wf_spec.task_specs.get(spec_name)
+        if self not in spec.inputs:
+            self.connect(spec)
+            spec.connect(self._wf_spec.end)
+        child = my_task._add_child(spec, TaskState.FUTURE)
+        child.triggered = True
+        child.task_spec._update(child)
+        return child.id
+
 
 class SimpleBpmnTask(BpmnTaskSpec):
     pass

SpiffWorkflow/bpmn/specs/defaults.py

@@ -33,6 +33,7 @@
     SubWorkflowTaskMixin,
     CallActivityMixin,
     TransactionSubprocessMixin,
+    EventSubprocessMixin,
     StartEventMixin,
     EndEventMixin,
     IntermediateCatchEventMixin,
@@ -88,6 +89,9 @@ class CallActivity(CallActivityMixin, BpmnSpecMixin):
 class TransactionSubprocess(TransactionSubprocessMixin, BpmnSpecMixin):
     pass
 
+class EventSubprocess(EventSubprocessMixin, BpmnSpecMixin):
+    pass
+
 class StartEvent(StartEventMixin, BpmnSpecMixin):
     pass
 

SpiffWorkflow/bpmn/specs/mixins/__init__.py

@@ -35,6 +35,7 @@
     SubWorkflowTask as SubWorkflowTaskMixin,
     CallActivity as CallActivityMixin,
     TransactionSubprocess as TransactionSubprocessMixin,
+    EventSubprocess as EventSubprocessMixin,
 )
 
 from .events.start_event import StartEvent as StartEventMixin
@@ -46,4 +47,4 @@
     EventBasedGateway as EventBasedGatewayMixin,
     SendTask as SendTaskMixin,
     ReceiveTask as ReceiveTaskMixin,
-)
+)

SpiffWorkflow/bpmn/specs/mixins/events/event_types.py

@@ -58,8 +58,12 @@ def _update_hook(self, my_task):
         elif my_task.state != TaskState.WAITING:
             my_task._set_state(TaskState.WAITING)
 
+        my_task.workflow.top_workflow.event_manager.add_task(my_task)
         self.event_definition.update_task(my_task)
 
+    def _on_ready_hook(self, my_task):
+        my_task.workflow.top_workflow.event_manager.remove_task(my_task)
+
     def _run_hook(self, my_task):
 
         self.event_definition.update_task_data(my_task)
@@ -72,6 +76,12 @@ def _predict_hook(self, my_task):
         if not isinstance(self.event_definition, CycleTimerEventDefinition):
             super()._predict_hook(my_task)
 
+    def _on_cancel(self, my_task):
+        for child in my_task:
+            child.workflow.top_workflow.event_manager.remove_task(child)
+        my_task.workflow.top_workflow.event_manager.remove_task(my_task)
+        super()._on_cancel(my_task)
+
 
 class ThrowingEvent(TaskSpec):
     """Base Task Spec for Throwing Event nodes."""

SpiffWorkflow/bpmn/specs/mixins/events/intermediate_event.py

@@ -54,9 +54,10 @@ def catches(self, my_task, event):
 class EventBasedGateway(CatchingEvent):
 
     def _predict_hook(self, my_task):
-        my_task._sync_children(self.outputs, state=TaskState.MAYBE)
+        my_task._sync_children(self.outputs, state=TaskState.WAITING)
 
     def _on_ready_hook(self, my_task):
         for child in my_task.children:
             if not child.internal_data.get('event_fired'):
                 child.cancel()
+

SpiffWorkflow/bpmn/specs/mixins/subworkflow_task.py

@@ -131,3 +131,7 @@ class TransactionSubprocess(SubWorkflowTask):
 
     def __init__(self, wf_spec, bpmn_id, subworkflow_spec, **kwargs):
         super().__init__(wf_spec, bpmn_id, subworkflow_spec, True, **kwargs)
+
+class EventSubprocess(SubWorkflowTask):
+    pass
+

SpiffWorkflow/bpmn/util/event.py

@@ -1,3 +1,5 @@
+from SpiffWorkflow.task import TaskState
+
 class BpmnEvent:
     def __init__(self, event_definition, payload=None, correlations=None, target=None):
         self.event_definition = event_definition
@@ -11,4 +13,58 @@ def __init__(self, name, event_type, value=None, correlations=None):
         self.name = name
         self.event_type = event_type
         self.value = value
-        self.correlations = correlations or {}
+        self.correlations = correlations or {}
+
+
+class EventManager:
+
+    def __init__(self, workflow):
+        self.workflow = workflow
+        self.tasks = {}
+
+    def add_task(self, my_task):
+        self.tasks[my_task.id] = my_task
+
+    def remove_task(self, my_task):
+        self.tasks.pop(my_task.id, None)
+
+    def get_waiting_tasks(self):
+        return [t.task_spec.event_definition.details(t) for t in self.tasks.values()]
+
+    def catch(self, event, internal=True):
+        if event.target is not None:
+            # This limits results to tasks in the specified workflow
+            tasks = [t for t in self.tasks.values() if t.workflow == event.target 
+                     and t.task_spec.event_definition.catches(t, event)]
+            if len(tasks) == 0:
+                event.target = event.target.parent_workflow
+                self.catch(event)
+        else:
+            self.update_collaboration(event)
+            tasks = [t for t in self.tasks.values() if t.task_spec.catches(t, event)]
+            # Figure out if we need to create an external event
+            if len(tasks) == 0 and internal:
+                self.workflow.bpmn_events.append(event)
+
+        for task in tasks:
+            task.task_spec.catch(task, event)
+            task.task_spec._update(task)
+
+        return len(tasks)
+
+    def update_collaboration(self, event):
+
+        def get_or_create_subprocess(task_spec, wf_spec):
+            for sp in self.workflow.subprocesses.values():
+                if sp.get_next_task(state=TaskState.WAITING, spec_name=task_spec) is not None:
+                    return sp
+            child_id = self.workflow.spec.start.trigger_wf(self.workflow.task_tree, wf_spec)
+            return self.workflow.subprocesses[child_id]
+
+        # Start a subprocess for known specs with start events that catch this
+        for name in self.workflow.spec.start.trigger_specs:
+            sp_spec = self.workflow.subprocess_specs.get(name)
+            for ts in sp_spec.bpmn_start_events:
+                if ts.event_definition == event.event_definition:
+                    subprocess = get_or_create_subprocess(ts.name, sp_spec.name)
+                    subprocess.correlations.update(event.correlations)