[enhancement]: MessageFlow part 2

rallen2010 commented 11 months ago

Feature description

In 23.1 we released MessageFlow Phase 1. This added:

BPMN Send Task
BPMN Receive Task
BPMN Message Catch Intermediate Event
BPMN Intermediate Throw Event
Basic Message Protocol
sendTask/messageThrow can send a message to another instance in the same Flows for APEX system, i.e., within the APEX Workspace.
in addition receiveTask / messageCatch can receive a message via the REST interface

Phase 2 would extend this to:

BPMN Work:

BPMN messageStartEvent
BPMN messageCatch Boundary Event on subProcesses and Call Activities (on Tasks?)
BPMN messageThrow End Event

Architecture Work:

Decouple sender and receivers using message queueing
Allow registration of endpoints
Routing of messages to other endpoint Flows systems

Describe alternatives you've considered

No response

Additional context

No response

rallen2010 commented 11 months ago

Existing Phase 1 spec for revision & extension...

Basic Functionality

SWe are building a declarative message passing mechanism that can be used with both task based elements (sendTask and receiveTask) and event-based elements (Intermediate Message Catch Event and Intermediate Message Throw Event).

Basic Concept

ReceiveTask generates a message name and a correlation key:value pair when the task starts. These, along with the task context (process id, subflow id, and step key) are inserted into a new run-time table, flow_subscriptions, and the subflow status is changed to waiting for message. An incoming message would supply a message name, correlation key and value, and optionally a payload (text, values, or a json snippet). If the message name and correlation key and value match a message name and correlation key-value in the subscriptions table, the stored task context is used to call into the Flows engine, transferring any payload into a process variable. After extracting values from the payload, the engine:

deletes the subscription
performs a step_complete
executes after-step variable expressions, and completes. If the payload contains json or some other formatted content, the After Task variable expressions can be used to extract values from this into Process Variables for use in later process steps.

Settings on the Receive Task / Message Catch Event definitions

Section: Message Name

Field: Expression Type. (. Static | ProcessVariable | SQL Statement | PL/SQL Expression | PL/SQL Function Body ) Field: Expression Value

(Note: This is a standard Flows for APEX "setting" and would be processes by flow_settings / flow_db_util

Section: Correlation Key

Field: Expression Type. (. Static | ProcessVariable | SQL Statement | PL/SQL Expression | PL/SQL Function Body ) Field: Expression Value

Section: Correlation Value

Field: Expression Type. (. Static | ProcessVariable | SQL Statement | PL/SQL Expression | PL/SQL Function Body ) Field: Expression Value

Payload Variable

Field: Payload Variable (Name of Process Variable to be created to hold payload - type CLOB).

bpmn:receiveTask BPMN Definition

MessageID and CorrelationKey would be stored inside bpmn:extensionElements as APEX extensions.

Does it make any sense to provide a 'execute PLSQL option here? I don't think the user could build the async catch event / callback receiver...

    <bpmn:receiveTask id="Activity_0fgm6pf" name="task name" apex:type="basicApexMessage" >
      <bpmn:extensionElements>
        <apex:messageName>
          <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
          <apex:expression></apex:expression>
        </apex:messageName>
        <apex:correlationKey>            
          <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
          <apex:expression></apex:expression>
        </apex::correlationKey>
        <apex:correlationValue>            
          <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
          <apex:expression></apex:expression>
        </apex::correlationValue>
        <apex:payloadVariable>myCLOB</apex:payloadVariable>
      </bpmn:extensionElements>
bpmn:intermediateCatchEvent - Message BPMN Definition

MessageID and CorrelationKey would be stored inside bpmn:extensionElements as APEX extensions.

Do these go inside the messageEventDefinition? Does it make any sense to provide a 'execute PLSQL option here? I don't think the user could build the async catch event / callback receiver...

    <bpmn:intermediateCatchEvent id="Activity_0fgm6pf" name="task name">
      <bpmn:messageEventDefinition id="MessageEventDefinition_091wxcp"  apex:type="basicApexMessage" />
        <bpmn:extensionElements>
          <apex:messageName>
            <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
            <apex:expression></apex:expression>
          </apex:messageName>
          <apex:correlationKey>            
            <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
            <apex:expression></apex:expression>
          </apex::correlationKey>
          <apex:correlationValue>            
            <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
            <apex:expression></apex:expression>
          </apex::correlationValue>
          <apex:payloadVariable>myCLOB</apex:payloadVariable>
        </bpmn:extensionElements>
      </bpmn:messageEventDefinition>
    </bpmn:intermediateCatchEvent>

Settings on the Send Task / Message Throw Event definitions

Section: Destination EndPoint

Field: Expression Type. (. Static | ProcessVariable | SQL Statement | PL/SQL Expression | PL/SQL Function Body ) Field: Expression Value

(Note: This is a standard Flows for APEX "setting" and would be processes by flow_settings / flow_db_util

Section: Message Name

as in Receive / Catch section above

Section: Correlation Key

as in Receive / Catch section above

Section: Correlation Value

as in Receive / Catch section above

Payload

Note this is different to Receive. This specifies how the Payload is created, not where to store it.

Field: Expression Type. (. Static | ProcessVariable | SQL Statement | PL/SQL Expression | PL/SQL Function Body ) Field: Expression Value

BPMN sendTask Definition

MessageID and CorrelationKey would be stored inside bpmn:extensionElements as APEX extensions.

<bpmn:sendTask id="Activity_0fgm6pf" name="task name" apex:type="basicApexMessage" 
  <bpmn:extensionElements>
    <apex:endPoint>
      <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
      <apex:expression></apex:expression>
    </apex:endPoint>
    <apex:messageName>
      <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
      <apex:expression></apex:expression>
    </apex:messageName>
    <apex:correlationKey>            
      <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
      <apex:expression></apex:expression>
    </apex::correlationKey>
    <apex:correlationValue>            
      <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
      <apex:expression></apex:expression>
    </apex::correlationValue>
    <apex:payload>
      <apex:expressionType>static|processVariable|sql|sqlmulti|expression|functionBody</apex:expressionType>
      <apex:expression></apex:expression>
    </apex:payload>
  </bpmn:extensionElements>

flow_subscriptions Table Definition (msub)

Column	Type	Constraints	Contents
msub_id	number	not null identity	generated
msub_name	varchar2(200)		required messageName
msub_key_name	varchar2(200)		required keyName
msub_key_value	varchar2(200)		required keyValue
msub_prcs_id	number	check FK to prcs_id delete cascade	prcs_id to callback
msub_sbfl_id	number	check valid subflow_id delete cascade	sbfl_id to callback
msub_step_key	varchar2(20)		step key to use on callback
msub_callback	varchar2(200)		callback type - allows tasks and messages to call back to different procedures
msub_callback_par	varchar2(200)		optional parameter if required for the callback, e.g., if a diagram contains 2 message startEvents, which to call
msub_created	timestamp with time zone

Correlator design

In first release, a message will be required to provide 100% match on messageName, key name and key value for the inbound message to correlate to a subscription.

In the event that a process needs to have alternate keys usable, multiple message Catch Events could be used with an Event Based Gateway, one to support each valid inbound message type.

for example, a buyer is often not sure if the supplier will quote the Buyer's PO number or the Supplier's Sales Order number when responding. A model could incorporate an intermediateCatchEvent for each of the two different message formats, preceded by an eventBasedGateway. Which ever of the two identifiers is used, the message will be received into the process instance. alternateMessageFormatsUsingEBG

REST interface

Phase 2 Implementation Tasks

Open Issues

mechanism for starting new instance using Message start Event (future release) - ensure compatible with table name for receiveTask subtype. executePlsql is no longer the correct type name for this. basicApex?? If a payload is present & you then want to extract elements from the JSON into separate process variables, do we implement that as a set of InMappings inside the task, or do we just store the JSON into a process variable and the user would then define a set of After Task variable expressions to extract the contents into process variables? A: We will leave this for the After Task variable expression set (receiveTask) or OnEvent variable expression set (message catch event). error handling

Phasing

Phase 1 - Release 23.1 - delivered

[x] sendTask - execute PL/SQL - already done for 23.1
[x] sendTask, receiveTask - basic declarative message send & receive - already done for 23.1
[x] message intermediateThrowEvent, CatchEvent - done
[x] eventBasedGateway support - done, inc ReceiveTask behind an EBG
[x] diagramming support for empty pools, and messageFlow between empty pool and a collaboration
[x] REST enabled - done

Phase 2 - Release 24.1

[x] messageStartEvent
[x] messageBoundaryEvent
[x] messageEndEvent

commi235 commented 10 months ago

Enhancement for Message Start Event (and probably other receivers):

[ ] Add an attribute to capture the entry point into the received JSON payload.
[ ] Add the possibility to set process variables based on the JSON payload taking the above entry point into account.

Example JSON:

{
  "message_name": "My Message",
  "correlation_key": "CustomerID",
  "correlation_value": "12345",
  "data": {
    "myKey": "ABC",
    "amount": 12345
  }
}

For this set entry point to "$.data" and then be able to set process variables using

myKeyVar -> $.myKey
amountVar -> $.amount

flowsforapex / apex-flowsforapex