Correlated Events and Complex Event Processing (CEP)

Simple Event Processing vs Complex Event Processing

Before we describe how AireFlow supports Complex Event Processing (CEP) it is worth understanding the difference between Simple and Complex event processing. In general most AireFlow workflows are likely to utilise Simple Event Processing more often than not, however there may be situations when you need complex event processing.

Simple Event Processing

In Simple Event Processing each event is handled independently, without considering its relationship with other events. In this model, an event might trigger a direct and immediate action or task.

For instance we might have a form submission event that triggers a notification task, or a review task.

Complex Event Processing

Complex Event Processing is about analysing multiple related events, often from different sources, to identify patterns or detect meaningful situations. The system observes and correlates events in real-time and then generates new (composite events) or triggers actions / tasks when all the complex event criteria are met.

For instance - in an HR system - we might correlate a low scoring happiness survey result, wellbeing day off requests and/or a request to move project into a higher level complex event indicating that the member of staff is unhappy and may need particular attention from the HR team.

Complex Event Processing with AireFlow

To achieve CEP on AireFlow we need to take advantage of AireFlow's ability to store state against a task as well as the ability to check and mutate that state.

Currently AireFlow does not support a dedicated CEP Task type - however we can use a Manual task for the purpose and simply filter out that task key from any UIs that show tasks to end users.

CEP Example

Suppose we wanted AireFlow to monitor a patient's National Early Warning Score - NEWS - a patient observation about the general condition of a patient typically in a hospital ward context. Whilst NEWS offers an instantanous score about the patients condition (the higher the score the worse the condition) of almost equal importance is the trend of successive NEWS scores - i.e. someone might have had a high NEWS score previously and whilst the most recent score is still high, it is better than previously thus indicating the patient is recovering. Equally, a patient that previously had a zero NEWS score might now have a low / medium NEWS score, indicating a deteriation that might require a clinical intervention.

To illustrate CEP in AireFlow we won't concern ourselve with a very complex (or indeed clinically accurate) algorithm as the approach will remain the same irrespective of the complexity.

For our example, we assume we will want to notify a NEWS surveillance team if the NEWS Score has increased since the previous reading - our worfklow might look like the following:

1. The surveilance task is instantiated by the event signalling that the patient has been admitted to the ward.
2. The surveilance task has a state variable previous-news-score (we intially set this to 100 to simplify the logic).
3. Any NEWS v2 form submission event will apply the mutation.
4. When the surveilance task is triggered (marked as complete) it sends a notification to the NEWS Surveilance team.

(As an improvement we would also want to retrigger the surveilance task once the Surveilance team has been notified, and we might also want to manage the observation period for NEWS - all of this is beyond the scope of this example).

To achieve the desired effect we use the following mutation, triggered by the submission of the NEWS v2 form:

[
  {% assign action = updateContext.Action | append: "" %}
  {% if action == "ClinicianFormSubmitted" %}
    {% assign previous-score = task.State.previous-news-score  | times: 1 %}
    {% assign new-score = updateContext.ValuesForExport.NEWSScore | times: 1 %}
    {% if previous-score < new-score %}
        {
            "op": "replace",
            "path": "/ActivityContext/Status",
            "value": "Completed"
        }
    {% else %}
        {
            "op": "replace",
            "path": "/ActivityContext/Status",
            "value": "InProgress"
        },
        {
            "op": "replace",
            "path": "/description",
            "value": "NEWS:  {{ new-score }} - recorded at {{ 'now' | date: '%Y-%b-%d %H:%M' }}"
        },
        {
            "op": "replace",
            "path": "/state/previous-news-score",
            "value": "{{ new-score }}"
        }
    {% endif %}
  {% endif %}
]

There is a fair bit to unpack here - so salient points are:

• We add a guard condition to make sure we only apply mutations when a form is submitted (rather than other events such as read). This is the {% if action == "ClinicianFormSubmitted" %} guard condition.
• We store the previous NEWS score and current NEWS score in liquid variables for ease of use. Note: we times the values by 1 (i.e. the ... | times: 1 ... part) to coerce the values into numbers for numeric comparison. Note also that the inital value for the state previous-news-score was set to 100 to make the logic simpler.
• The line {% if previous-score < new-score %} determines whether the latest score is higher than previously and thus (in our simple scenario) meets the criteria of a deteriorating patient. In this case we mark the task status as Completed and the subsequent notification task is then triggered.
• If the patient is not deemed to be deteriorating we overwrite the value for the previous-news-score state variable, and update the description so useful information is displayed to end users if the task is visible on the UI, as well as marking the task as InProgress.

See below for a screenshot of the task with all the aspects described above: