added a couple models for sending and receiving messages

migrations/versions/a5a93fe63899_.py

@@ -1,8 +1,8 @@
 """empty message
 
-Revision ID: b4f678040235
+Revision ID: a5a93fe63899
 Revises: 
-Create Date: 2022-07-25 09:46:39.406847
+Create Date: 2022-07-27 08:51:44.791339
 
 """
 from alembic import op
@@ -10,7 +10,7 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision = 'b4f678040235'
+revision = 'a5a93fe63899'
 down_revision = None
 branch_labels = None
 depends_on = None
@@ -131,6 +131,15 @@ def upgrade():
     sa.ForeignKeyConstraint(['user_uid'], ['user.uid'], ),
     sa.PrimaryKeyConstraint('id')
     )
+    op.create_table('queued_send_message',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('process_instance_id', sa.Integer(), nullable=False),
+    sa.Column('bpmn_element_id', sa.String(length=50), nullable=False),
+    sa.Column('correlation_name', sa.String(length=50), nullable=False),
+    sa.Column('correlation_value', sa.String(length=50), nullable=False),
+    sa.ForeignKeyConstraint(['process_instance_id'], ['process_instance.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
     op.create_table('task_event',
     sa.Column('id', sa.Integer(), nullable=False),
     sa.Column('user_id', sa.Integer(), nullable=False),
@@ -173,6 +182,7 @@ def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
     op.drop_table('data_store')
     op.drop_table('task_event')
+    op.drop_table('queued_send_message')
     op.drop_table('file')
     op.drop_table('active_task')
     op.drop_table('user_group_assignment')

src/spiffworkflow_backend/models/queued_receive_message.py

@@ -0,0 +1,21 @@
+"""Principal."""
+from dataclasses import dataclass
+
+from flask_bpmn.models.db import db
+from flask_bpmn.models.db import SpiffworkflowBaseDBModel
+from sqlalchemy import ForeignKey
+
+from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
+
+
+@dataclass
+class QueuedReceiveMessageModel(SpiffworkflowBaseDBModel):
+    """Messages from a process instance that are ready to receive a message from a task."""
+
+    __tablename__ = "queued_receive_message"
+
+    id = db.Column(db.Integer, primary_key=True)
+    process_instance_id = db.Column(ForeignKey(ProcessInstanceModel.id), nullable=False)  # type: ignore
+    bpmn_element_id = db.Column(db.String(50), nullable=False)
+    correlation_name = db.Column(db.String(50), nullable=False)
+    correlation_value = db.Column(db.String(50), nullable=False)

src/spiffworkflow_backend/models/queued_send_message.py

@@ -0,0 +1,21 @@
+"""Principal."""
+from dataclasses import dataclass
+
+from flask_bpmn.models.db import db
+from flask_bpmn.models.db import SpiffworkflowBaseDBModel
+from sqlalchemy import ForeignKey
+
+from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
+
+
+@dataclass
+class QueuedSendMessageModel(SpiffworkflowBaseDBModel):
+    """Messages from a process instance that are ready to send to a receiving task."""
+
+    __tablename__ = "queued_send_message"
+
+    id = db.Column(db.Integer, primary_key=True)
+    process_instance_id = db.Column(ForeignKey(ProcessInstanceModel.id), nullable=False)  # type: ignore
+    bpmn_element_id = db.Column(db.String(50), nullable=False)
+    correlation_name = db.Column(db.String(50), nullable=False)
+    correlation_value = db.Column(db.String(50), nullable=False)

src/spiffworkflow_backend/services/process_instance_processor.py

@@ -39,6 +39,7 @@ from spiffworkflow_backend.models.principal import PrincipalModel
 from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
 from spiffworkflow_backend.models.process_instance import ProcessInstanceStatus
 from spiffworkflow_backend.models.process_model import ProcessModelInfo
+from spiffworkflow_backend.models.queued_send_message import QueuedSendMessageModel
 from spiffworkflow_backend.models.task_event import TaskAction
 from spiffworkflow_backend.models.task_event import TaskEventModel
 from spiffworkflow_backend.models.user import UserModelSchema
@@ -448,6 +449,16 @@ class ProcessInstanceProcessor:
         try:
             self.bpmn_process_instance.refresh_waiting_tasks()
             self.bpmn_process_instance.do_engine_steps(exit_at=exit_at)
+            # NOTE: MESSAGE - should we check for thrown_events here?
+            if self.bpmn_process_instance.thrown_events:
+                for thrown_event in self.bpmn_process_instance.thrown_events:
+                    queued_message = QueuedSendMessageModel(
+                        process_instance_id=self.process_instance_model.id,
+                        bpmn_element_id=thrown_event.task_name,
+                    )
+                    db.session.add(queued_message)
+                    db.session.commit()
+
         except WorkflowTaskExecException as we:
             raise ApiError.from_workflow_exception("task_error", str(we), we) from we