Activiti: Signals between process definitions in cloud

[ryandawsonuk]

We want to be able to throw and catch signals across process definitions. But what if the signals are for process definitions that are in different runtime bundles?

It would make sense for the throwing runtime bundle to put a signal on a shared queue and for other runtime bundles that might need to consume signals to be listening on that queue.

More specifically runtime bundles are already able to listen on a shared Spring Cloud Stream channel using the "integrationResultsConsumer" - https://github.com/Activiti/activiti-cloud-runtime-bundle-service/blob/9bf1de8bc63e4a2ff0f85edfbda120b99a0e5181/activiti-cloud-services-runtime-bundle/activiti-cloud-services-connectors/src/main/java/org/activiti/services/connectors/channel/ProcessEngineIntegrationChannels.java#L26

For example the runtime bundle can listen to it like
https://github.com/Activiti/activiti-cloud-runtime-bundle-service/blob/9bf1de8bc63e4a2ff0f85edfbda120b99a0e5181/activiti-cloud-starter-runtime-bundle/src/test/java/org/activiti/cloud/starter/tests/runtime/ServiceTaskConsumerHandler.java#L38

and it can send to it through MessageProducerCommandContextCloseListener

We need a way to send signals to a stream when the runtime bundle reaches a signal throw in the process definition (it should be property-configurable whether we send to stream or just works as it does currently). We also need a way to handle the inbound signals on the receiving runtime bundle.

This means that at startup the receiving runtime bundle is going to need contain a StreamListener so that it will to start listening on the queue. But it will have to be created at startup time as it won't know in advance what process definition it contains and so it won't know what signals to listen for. This would mean that the "condition = " part of the StreamListener would have to be dynamic. I think this can be any spring expression so we should be able to do that.

Presumably this will require some changes to IntermediateThrowSignalEventActivityBehavior.

We also hope to do this for messages - have created a separate ticket for that (https://github.com/Activiti/Activiti/issues/1691).

[ryandawsonuk]

Either this issue or #1691 can be tackled first. The order in which we approach them does not matter.

[daisuke-yoshimoto]

Investigation

Current signal behavior

Catching a Signal Event(IntermediateCatchSignalEventActivityBehavior)

• https://github.com/Activiti/Activiti/blob/develop/activiti-engine/src/main/java/org/activiti/engine/impl/bpmn/behavior/IntermediateCatchSignalEventActivityBehavior.java

• Add a record to the event subscription table(act_ru_event_subscr) for each signal.

commandContext.getEventSubscriptionEntityManager().insertSignalEvent(signalName, signal, executionEntity);

Throwing a Signal Event(IntermediateThrowSignalEventActivityBehavior)

• https://github.com/Activiti/Activiti/blob/develop/activiti-engine/src/main/java/org/activiti/engine/impl/bpmn/behavior/IntermediateThrowSignalEventActivityBehavior.java

1. Querying for Signal Event subscriptions

   • If process instance scope, search under the condition(processInstanceId and signalName)
   • If global scope, search under the condition(only signalName)

2. Execute acquired subscriptions serially.

   • If async, execute a subscription by AsyncExecutor.

EventSubscriptionEntityManager eventSubscriptionEntityManager = commandContext.getEventSubscriptionEntityManager();
List<SignalEventSubscriptionEntity> subscriptionEntities = null;
if (processInstanceScope) {
    subscriptionEntities = eventSubscriptionEntityManager
    .findSignalEventSubscriptionsByProcessInstanceAndEventName(execution.getProcessInstanceId(), eventSubscriptionName);
} else {
    subscriptionEntities = eventSubscriptionEntityManager.findSignalEventSubscriptionsByEventName(eventSubscriptionName, execution.getTenantId());
}

for (SignalEventSubscriptionEntity signalEventSubscriptionEntity : subscriptionEntities) {
    Context.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
                    ActivitiEventBuilder.createSignalEvent(ActivitiEventType.ACTIVITY_SIGNALED,
                                                           signalEventSubscriptionEntity.getActivityId(),
                                                           eventSubscriptionName,
                                                           null,
                                                           signalEventSubscriptionEntity.getExecutionId(),
                                                           signalEventSubscriptionEntity.getProcessInstanceId(),
                                                           signalEventSubscriptionEntity.getProcessDefinitionId()));

   eventSubscriptionEntityManager.eventReceived(signalEventSubscriptionEntity,
                                                         null,
                                                         signalEventDefinition.isAsync());
}

New Behavior

Catching a Signal Event(IntermediateCatchSignalEventActivityBehavior)

• Same as the current Behavior.

Throwing a Signal Event(IntermediateThrowSignalEventActivityBehavior)

• Like MQServiceTaskBehavior,after transaction commit, send message by Spring Cloud Stream by each signal subscription.
  
  
  
  • This transaction management is wrong.
  
  
*

[igdianov]

@daisuke-yoshimoto @ryandawsonuk I suggest to review existing Activiti signal event test cases and problems in order to understand the full scope of work in perspective.

There are two scopes for signal events in Activiti: global and processInstance. Throw signal event behavior can be executed either synchronously in the current transaction or scheduled asynchronously provided by signal definition with activiti:async attribute.

Signal catch events MUST subscribe before throw signal is emitted. Many existing test cases expect to receive event signals synchronously in the current transaction by subscribing in the previous transaction.

There exists race conditions between throw/catch events in the same transaction scope due to non-deterministic handling of execution flow.

Catch signals behaviors also cannot resolve signal events thrown in previous transactions of the process instance because throw signal event behavior life-time is limited to a single transaction scope depending on sync or async execution.

In other words, there is a mismatch between signal definitions and behaviors with current implementation. I am open to collaborate on this.

[salaboy]

@igdianov @daisuke-yoshimoto great start on this.
My two cents on the subject:

If we introduce Spring Cloud Streams to deal with global and processInstance signals, this will be only for Activiti Cloud, meaning that a new behaviour can be defined. We don't need to replace the current Activiti behaviour, but override it for cloud deployments (in the same way that we did with ServiceTasks for RuntimeBundles). In cloud, the Sync option is not longer needed (IMO), for that reason the new behaviour should consider only async (and more reactive) scenarios.
If we try to cover all the test cases in the Activiti core engine we will fail to implement a cloud solution where the async behaviour is a expected.

[cijujoseph]

As discussed in the meeting, I tried to write down the signal use cases and ideas. Not sure if this is the best way to put it down. Hopefully this will be of some help for the cloud implementation

Global Scoped Signal

When a signal with global scope is thrown:

• A process can be started
• An event sub process can be started
• Boundary signal catch component in a BPMN model can be triggered
• Intermediate catch component in a BPMN model can be triggered

Signal delivery use cases:

• Delivery only to active customers. Expire the signal after delivery.
• Delivery to inactive consumers as well (those who previously registered for this signal but currently down -> signal thrown time > consumer subscription time). This means signals must be retained unitl delivered to all subscribed customers.
• Delivery to new consumers who registers themselves after the signal is thrown but within certain time-frame(signal ttl) (signal thrown time < consumer subscription time)(). This means signals must be retained until signal ttl.

Process Instance Scoped Signal

When a signal with process instance scope is thrown:

• An event sub process can be started
• Boundary signal catch component in a BPMN model can be triggered
• Intermediate catch component in a BPMN model can be triggered

Signal delivery use cases:

• Delivery only to active consumers - when producer send and consumer creation happens in different transactions & consumer is created first!
• Delivery to inactive consumers - when producer send and consumer creation happens in different transactions & consumer registered first
• Delivery to new & existing consumers when send and receive happens in the same transaction scope

How can a signal be thrown?

Simple answer is that it is via signal API. If you drill it further down, it can be split to these scenarios:

• Via API both Java & Rest
• From BPMN model via signal throw component
• From an event listener attached to a process model using the re-throw option. eg:

Finally some solution thoughts...

A couple of additional extensions such as durable/non-durable, time to live etc may cover the above use cases from a modelling standpoint. From an implementation point of view, hopefully those are supported by spring-cloud-stream implementations of rabbitmq and others....

[salaboy]

Initial implementation merged, closing and creating a new one for adding more features.