События-сообщения (Message Events)

Этот раздел перенесён из документации Camunda 7 и в дальнейшем будет доработан с учётом особенностей OpenBPM Engine

События-сообщения — это события, которые ссылаются на именованное сообщение. Сообщение имеет имя и полезную нагрузку (payload). В отличие от сигнала, событие-сообщение всегда направлено конкретному получателю.

Определение сообщения

Определение события-сообщения объявляется с помощью элемента messageEventDefinition. Атрибут messageRef ссылается на элемент message, объявленный как дочерний элемент корневого элемента definitions.

Ниже приведён фрагмент процесса, в котором объявлены два события-сообщения, используемые в стартовом событии и промежуточном перехватывающем событии-сообщении.

Пример

<definitions id="definitions"
  xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
  xmlns:camunda="http://activiti.org/bpmn"
  targetNamespace="Examples"
  xmlns:tns="Examples">

  <message id="newInvoice" name="newInvoiceMessage" />
  <message id="payment" name="paymentMessage" />

  <process id="invoiceProcess">

    <startEvent id="messageStart" >
        <messageEventDefinition messageRef="newInvoice" />
    </startEvent>
    ...
    <intermediateCatchEvent id="paymentEvt" >
        <messageEventDefinition messageRef="payment" />
    </intermediateCatchEvent>
    ...
  </process>
</definitions>

Выражения

Вы можете использовать выражения для имени сообщения в определении события-сообщения (за исключением стартового события процесса). Имя сообщения вычисляется в момент, когда процесс достигает области действия события. Например, когда экземпляр процесса доходит до промежуточного перехватывающего события-сообщения, выражение в имени сообщения будет разрешено.

Используя выражения в имени сообщения, можно динамически изменять имя сообщения на основе переменных процесса. Пример:

<message id="newInvoice" name="newInvoiceMessage-${execution.processBusinessKey}" />

Примечание: Использование выражений в имени сообщения стартового события определения процесса запрещено. Если использовать выражение в определении сообщения и сослаться на него в стартовом событии процесса верхнего уровня, это приведёт к ошибке.

Однако использование выражений разрешено в стартовом событии подпроцесса. Соответственно, ссылка на сообщение с выражением в имени будет корректно работать, если стартовое событие находится внутри подпроцесса.

Расширения OpenBPM Engine

Атрибуты

camunda:asyncBefore, camunda:asyncAfter, camunda:exclusive, camunda:jobPriority

Элементы расширений

camunda:inputOutput

Ограничения

–

Message API

Как встраиваемый процессный движок, OpenBPM Engine не занимается приёмом сообщений. Это зависит от окружения и включает платформо-специфичные действия, такие как подключение к очередям или топикам JMS или обработка Web Service / REST-запросов.

Таким образом, приём сообщений должен быть реализован в приложении или инфраструктуре, в которую встроен процессный движок.

После получения сообщения вы можете либо использовать встроенный механизм корреляции, либо явно доставить сообщение для запуска нового экземпляра процесса или продолжения ожидающего выполнения.

Использование методов корреляции RuntimeService

Движок предоставляет базовый механизм корреляции, который либо сигнализирует выполнению, ожидающему конкретное сообщение, либо создаёт экземпляр процесса со стартовым событием-сообщением.

RuntimeService предоставляет fluent API для корреляции сообщений.

Результатом корреляции является объект типа MessageCorrelationResult. Он содержит тип корреляции:

  • execution — если сообщение было сопоставлено с промежуточным перехватывающим событием,

  • processDefinition — если сообщение было сопоставлено со стартовым событием.

Если тип равен execution, результат содержит объект Execution, доступный через метод result.getExecution().

Если тип равен processDefinition, результат содержит объект ProcessInstance, созданный через стартовое событие, доступный через result.getProcessInstance().

// correlate the message
MessageCorrelationResult result = runtimeService.createMessageCorrelation("messageName")
  .processInstanceBusinessKey("AB-123")
  .setVariable("payment_type", "creditCard")
  .correlateWithResult();

Также можно явно найти подписку и активировать её:

ProcessInstance pi = runtimeService.startProcessInstanceByKey("processWaitingInReceiveTask");

EventSubscription subscription = runtimeService.createEventSubscriptionQuery()
  .processInstanceId(pi.getId())
  .eventType("message")
  .singleResult();

runtimeService.messageEventReceived(
  subscription.getEventName(),
  subscription.getExecutionId()
);

messageName идентифицирует сообщение так, как оно задано в атрибуте name элемента message в BPMN XML.

Корреляция считается успешной, если существует ровно один подходящий объект из следующих:

  • Определение процесса — если процесс может быть запущен сообщением с именем messageName

  • Выполнение (экземпляр процесса) — если выполнение ожидает сообщение с именем messageName (если оно указано) и экземпляр процесса соответствует заданным businessKey и correlationKeys (если указаны)

correlationKeys сопоставляются с переменными экземпляра процесса. Если messageName не задано, подойдёт любое выполнение, соответствующее остальным критериям. Это полезно, если отправляющая сторона знает только динамический ключ корреляции, но не знает имя сообщения, заданное в модели процесса.

Также возможно за один вызов сопоставить сообщение с несколькими выполнениями и одновременно запустить процесс:

List<MessageCorrelationResult> results = runtimeService
  .createMessageCorrelation("aMessageName")
  .correlateAllWithResult();

Результатом будет список объектов MessageCorrelationResult, каждый из которых соответствует одной корреляции.

Можно получить переменные процесса в момент корреляции сообщения. Параметр shouldDeserializeValues определяет, должны ли значения переменных быть десериализованы:

MessageCorrelationResultWithVariables result = runtimeService
  .createMessageCorrelation("aMessageName")
  .setVariable("name", "value")
  .correlateWithResultAndVariables(shouldDeserializeValues);

VariableMap processVariables = result.getVariables();

Кроме того, builder корреляции сообщений позволяет сопоставлять сообщение по локальным переменным выполнения:

List<MessageCorrelationResult> results = runtimeService
  .createMessageCorrelation("aMessageName")
  .localVariableEquals("localVarName", "localVarValue")
  .correlateAllWithResult();

В этом случае подходящее выполнение выбирается на основе переменных, существующих в его собственной области видимости, без учёта родительских областей.

При успешной корреляции связанный или вновь созданный экземпляр процесса обновляется переменными из карты processVariables.

Явная отправка сообщения (Explicitly Triggering a Message)

В качестве альтернативы вы можете явно доставить сообщение, чтобы запустить новый экземпляр процесса или продолжить выполнение, ожидающее сообщение.

Если сообщение должно инициировать запуск нового экземпляра процесса, вы можете использовать API корреляции сообщений:

ProcessInstance startedProcessInstance = runtimeService
  .createMessageCorrelation("messageName")
  .processInstanceBusinessKey("businessKey")
  .setVariable("name", "value")
  .correlateStartMessage();

// or

MessageCorrelationResultWithVariables result = runtimeService
  .createMessageCorrelation("aMessageName")
  .processInstanceBusinessKey("businessKey")
  .startMessageOnly()
  .setVariable("name", "value")
  .correlateWithResultAndVariables(shouldDeserializeValues);

ProcessInstance startedProcessInstance = result.getProcessInstance();
VariableMap processVariables = result.getVariables();

Также можно воспользоваться одним из следующих методов RuntimeService:

ProcessInstance startProcessInstanceByMessage(String messageName);
ProcessInstance startProcessInstanceByMessage(String messageName, Map<String, Object> processVariables);
ProcessInstance startProcessInstanceByMessage(String messageName, String businessKey, Map<String, Object> processVariables);

Эти методы позволяют запустить экземпляр процесса, используя указанное сообщение.

Если сообщение должно быть получено уже существующим экземпляром процесса, сначала необходимо выполнить корреляцию сообщения с конкретным экземпляром процесса (см. следующий раздел), а затем инициировать продолжение ожидающего выполнения.

RuntimeService предоставляет следующие методы для запуска выполнения на основе подписки на событие-сообщение:

void messageEventReceived(String messageName, String executionId);
void messageEventReceived(String messageName, String executionId, HashMap<String, Object> processVariables);

Для асинхронной корреляции сообщений с существующими экземплярами процессов можно использовать Batch-операции.

Поиск подписок на события-сообщения

Движок поддерживает стартовые события-сообщения и промежуточные события-сообщения.

В случае стартового события-сообщения подписка связана с конкретным определением процесса. Такие подписки можно получить с помощью ProcessDefinitionQuery:

ProcessDefinition processDefinition = repositoryService
  .createProcessDefinitionQuery()
  .messageEventSubscription("newCallCenterBooking")
  .singleResult();

Поскольку для одного сообщения может существовать не более одного определения процесса, запрос всегда возвращает либо ноль, либо один результат.

Если определение процесса обновляется, подписки на сообщения предыдущей версии отменяются. Это также относится к событиям-сообщениям, которые отсутствуют в новой версии процесса.

В случае промежуточного перехватывающего события-сообщения подписка связана с конкретным выполнением (Execution). Такие подписки можно искать с помощью ExecutionQuery:

Execution execution = runtimeService
  .createExecutionQuery()
  .messageEventSubscriptionName("paymentReceived")
  .processVariableValueEquals("orderId", message.getOrderId())
  .singleResult();

Подобные запросы называются корреляционными запросами и, как правило, требуют знания особенностей процесса (в данном примере предполагается, что существует не более одного экземпляра процесса для заданного orderId).

Стартовое событие-сообщение (Message Start Event)

Стартовое событие-сообщение используется для запуска экземпляра процесса с помощью именованного сообщения. Это позволяет выбрать корректное стартовое событие из набора альтернативных стартовых событий на основе имени сообщения.

При развертывании определения процесса с одним или несколькими стартовыми событиями-сообщениями необходимо учитывать следующее:

  • Имя стартового события-сообщения должно быть уникальным в рамках одного определения процесса. Если два или более стартовых события ссылаются на одно и то же сообщение или на сообщения с одинаковыми именами, движок выбросит исключение при деплое.

  • Имя стартового события-сообщения должно быть уникальным среди всех развернутых определений процессов. Если сообщение с таким именем уже используется в другом процессе, деплой завершится ошибкой.

  • Версионирование процессов: при развертывании новой версии процесса подписки на сообщения предыдущей версии отменяются. Это также касается событий-сообщений, отсутствующих в новой версии.

Запуск процесса через стартовое событие-сообщение выполняется следующими методами RuntimeService:

ProcessInstance startProcessInstanceByMessage(String messageName);
ProcessInstance startProcessInstanceByMessage(String messageName, Map<String, Object> processVariables);
ProcessInstance startProcessInstanceByMessage(String messageName, String businessKey, Map<String, Object> processVariables);

Параметр messageName соответствует значению атрибута name элемента message, на который ссылается messageRef в элементе messageEventDefinition.

При запуске процесса действуют следующие правила:

  • Стартовые события-сообщения поддерживаются только в процессах верхнего уровня. Во встроенных подпроцессах они не поддерживаются.

  • Если процесс содержит несколько стартовых событий-сообщений, метод runtimeService.startProcessInstanceByMessage(…​) позволяет выбрать нужное стартовое событие.

  • Если процесс содержит несколько стартовых событий-сообщений и одно none-стартовое событие, методы startProcessInstanceByKey(…​) и startProcessInstanceById(…​) запускают процесс через none-стартовое событие.

  • Если процесс содержит несколько стартовых событий-сообщений и не содержит none-стартового события, методы startProcessInstanceByKey(…​) и startProcessInstanceById(…​) выбрасывают исключение.

  • Если процесс содержит одно стартовое событие-сообщение, методы startProcessInstanceByKey(…​) и startProcessInstanceById(…​) запускают процесс через это событие.

  • Если процесс запускается из Call Activity, стартовые события-сообщения поддерживаются только если:

    • помимо стартовых событий-сообщений в процессе есть одно none-стартовое событие, либо

    • процесс содержит ровно одно стартовое событие-сообщение и не содержит других стартовых событий.

XML-представление стартового события-сообщения — это обычное объявление стартового события с дочерним элементом messageEventDefinition:

<definitions id="definitions"
  xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL"
  xmlns:camunda="http://activiti.org/bpmn"
  targetNamespace="Examples"
  xmlns:tns="Examples">

  <message id="newInvoice" name="newInvoiceMessage" />

  <process id="invoiceProcess">

    <startEvent id="messageStart">
      <messageEventDefinition messageRef="tns:newInvoice" />
    </startEvent>
    ...
  </process>

</definitions>

Процесс может быть запущен с помощью одного из нескольких различных сообщений. Это полезно, если процесс должен по-разному реагировать на разные стартовые события, но затем продолжаться по единому сценарию.

event message start alternative

Промежуточное перехватывающее сообщение (Message Intermediate Catching Event)

Когда токен достигает промежуточного перехватывающего события сообщения, он останавливается и ожидает поступления сообщения с соответствующим именем. Как уже было описано ранее, сообщение должно быть передано в движок с помощью соответствующих вызовов API.

Следующий пример демонстрирует различные события сообщений в модели процесса:

event message

<intermediateCatchEvent id="message">
  <messageEventDefinition messageRef="newCustomerMessage" />
</intermediateCatchEvent>

Вместо промежуточного перехватывающего события сообщения можно рассмотреть использование Receive Task, который может использоваться для схожих целей, но при этом поддерживает использование совместно с граничными событиями (boundary events). В сочетании с промежуточным перехватывающим событием сообщения также может использоваться шлюз, основанный на событиях (Event-based Gateway).

Граничное событие сообщения (Message Boundary Event)

Граничные события являются перехватывающими событиями, которые прикрепляются к активности. Это означает, что пока активность выполняется, граничное событие сообщения ожидает поступления сообщения с заданным именем. В зависимости от конфигурации граничного события, при получении сообщения возможны два варианта поведения:

  • Прерывающее граничное событие: активность прерывается, и выполнение продолжается по исходящему из события потоку управления.

  • Непрерывающее граничное событие: один токен остаётся в активности, а дополнительный токен создаётся и следует по исходящему из события потоку управления.

Промежуточное генерирующее сообщение (Message Intermediate Throwing Event)

Промежуточное генерирующее событие сообщения отправляет сообщение во внешнюю систему. Данное событие имеет то же поведение, что и Service Task.

event message throwing

<intermediateThrowEvent id="message">
  <messageEventDefinition camunda:class="io.openbpm.bpm.MyMessageServiceDelegate" />
</intermediateThrowEvent>

Расширения OpenBPM Engine для messageEventDefinition

Атрибуты

camunda:class, camunda:delegateExpression, camunda:expression, camunda:resultVariable, camunda:topic, camunda:type, camunda:taskPriority

Элементы расширений

camunda:field, camunda:connector

Ограничения

Один из атрибутов camunda:class, camunda:delegateExpression, camunda:type или camunda:expression является обязательным

Атрибут camunda:resultVariable может использоваться только в сочетании с атрибутом camunda:expression

Атрибут camunda:type может принимать только значение external

Атрибут camunda:topic может использоваться только в случае, если атрибут camunda:type установлен в значение external

Атрибут camunda:taskPriority может использоваться только в случае, если атрибут camunda:type установлен в значение external

Завершающее событие сообщения (Message End Event)

Когда выполнение процесса достигает завершающего события сообщения, текущий путь выполнения завершается и отправляется сообщение. Завершающее событие сообщения имеет то же поведение, что и Service Task.

<endEvent id="end">
  <messageEventDefinition camunda:class="io.openbpm.bpm.MyMessageServiceDelegate" />
</endEvent>

Расширения OpenBPM Engine для messageEventDefinition

Атрибуты

camunda:class, camunda:delegateExpression, camunda:expression, camunda:resultVariable, camunda:topic, camunda:type, camunda:taskPriority

Элементы расширений

camunda:field, camunda:connector

Ограничения

Один из атрибутов camunda:class, camunda:delegateExpression, camunda:type или camunda:expression является обязательным

Атрибут camunda:resultVariable может использоваться только в сочетании с атрибутом camunda:expression

Атрибут camunda:type может принимать только значение external

Атрибут camunda:topic может использоваться только в случае, если атрибут camunda:type установлен в значение external

Атрибут camunda:taskPriority может использоваться только в случае, если атрибут camunda:type установлен в значение external

Лицензия и атрибуция

Эта документация была создана на базе материала "Camunda 7 Docs" от Camunda, находится под лицензией Creative Commons Attribution-ShareAlike 3.0 Unported License .

Оригинал документации: https://docs.camunda.org