Module khepri_evf

Description
Data Types
Function Index
Function Details

Khepri event filters.

Description

Khepri event filters.

Triggers allow a user to associate an event, described by an event filter, to an action. When an event matches the registered event filter, the associated action is evaluated.

Events and event filters

Changes to the tree

When a tree node is created, modified or deleted, a tree change is emitted.

To register a trigger that targets tree changes, you can create a tree-change event filter explicitly:

EventFilter = khepri_evf:tree([stock, wood, <<"oak">>], %% Required
                              #{on_actions => [delete], %% Optional
                                priority => 10}),       %% Optional
 
ok = khepri:register_trigger(
       StoreId,
       TriggerId,
       EventFilter,
       Action).

It's possible to pass a path pattern directly as an event filter: this is the same as creating an explicit tree-change event filter with default options.

EventFilter = "/:stock/:wood/oak",
 
ok = khepri:register_trigger(
       StoreId,
       TriggerId,
       EventFilter,
       Action).

The path pattern can match many paths. The monitored paths don't neew to exist when the trigger is registered.

Termination of a process

Khepri can monitor a process. When this process exits, a process event is emitted.

To register a trigger that targets process termination, you can create a process event filter explicitly:

EventFilter = khepri_evf:process(self(),             %% Required
                                 #{priority => 10}), %% Optional
 
ok = khepri:register_trigger(
       StoreId,
       TriggerId,
       EventFilter,
       Action).

It's possible to pass a PID directly as an event filter: this is the same as creating an explicit process event filter with default options.

EventFilter = self(),
 
ok = khepri:register_trigger(
       StoreId,
       TriggerId,
       EventFilter,
       Action).

The Khepri leader monitors the given process. When the process exits or if it's not running in the first place, it will trigger the execution of the trigger action.

If the monitored process is on a different Erlang node than the Ra leader, the loss of connection between the two nodes (i.e. the noconnection exit reason) is handled in a specific way:

If the remote node is a member of the Khepri cluster, Khepri will start to monitor that node after receiving the noconnection exit reaison. Then two things can happen:
- The node comes back online. In this case, processes that are supposed to run on it are monitored again. If the node was restarted and the processes were killed, this new monitoring will detect this situation.
- The node is removed from the cluster. In this case, it is unlikely to come back. Thus, the processes that were running on it are considered as dead.
If the remote node is not a member of the Khepri cluster, Khepri will consider that the processes supposed to run on this node are dead, even if this is network partition and the node and its processes could be reachable again in the future.

Actions

Trigger descriptor

The trigger descriptor (khepri_event_handler:trigger_descriptor()) is a record used to encapsulate the properties of a specific triggered trigger: it contains details about:

the originating store and trigger IDs
the originating event
the action properties

This trigger descriptor is passed to the action and allows the action to distinguish each instance of a trigger.

=== Execution of a stored procedure

The action takes be a path to a stored procedure. The action takes the form of:

a path to a stored procedure, or
a {sproc, StoredProcPath} tuple

The stored procedure must take a single argument, the trigger descriptor.

The stored procedure does not have to exist when the trigger is registered. If the trigger is triggered while the stored procedure is missing, the event will be ignored.

The return value of the function is ignored.

Here is an example of a function that can be used as a stored procedure:

my_stored_procedure(#khepri_trigger{...}) ->
    do_something_with_event().

Execution of an arbitrary MFA

The action takes the form of:

a {Module, Function, ArgsList} tuple, or
a {apply, {Module, Function, ArgsList}} tuple

When the action is triggered, the MFA is executed with the given arguments list with the trigger descriptor appended to it.

The return value of the function is ignored.

Here is an example of a function that can be used with {my_mod, my_func, [ExtraArg]}:

-module(my_mod).
-export([my_func/2]).
 
my_func(ExtraArg, #khepri_trigger{...}) ->
    do_something_with_event(ExtraArg).

Send of a message

The action takes the form of:

a PID, or
a {send, Pid, Priv} tuple

When the action is triggered, the trigger descriptor is sent as a message to the given PID. The Priv term is added to the action properties. If the action is the PID alone, Priv defaults to undefined and nothing is added to the action properties.

The target process is not monitored. If it's not running anymore or it is on a node that is unreachable at the time of the event, the event is triggered.

Here is an example of a receive block that expects a trigger descriptor:

receive
    #khepri_trigger{...} ->
        do_something_with_event()
end.

Where is the action evaluated?

By default, the action is evaluated by the Khepri leader.

It's possible to indicate where to evaluate the trigger when the trigger is registered, using the where option (see khepri:trigger_options()).

Here is an example of a trigger where the action will be evaluated on all members of the Khepri cluster at the time of the event.

EventFilter = khepri_evf:tree([stock, wood, <<"oak">>], %% Required
                              #{on_actions => [delete], %% Optional
                                priority => 10}),       %% Optional
 
ok = khepri:register_trigger(
       StoreId,
       TriggerId,
       EventFilter,
       Action,
       #{where => all_members}).

Data Types

event()

event() = tree_event() | process_event()

An record representing an event.

event_filter()

event_filter() = tree_event_filter() | process_event_filter()

An event filter.

The following event filters are supported:

Tree event filter (tree_event_filter()
Process event filter (process_event_filter()

An event filter can be explicitly constructed using the functions provided in this module. However, some common types will be automatically detected and converted to an event filter with default properties. See each event filter type for more details.

event_filter_or_compat()

event_filter_or_compat() = khepri_evf:event_filter() | khepri_path:pattern() | pid()

An event filter, or any type that can be automatically converted to an event filter.

For instance, a Khepri path or a string can be converted to a tree_event_filter().

priority()

priority() = integer()

An event filter priority.

This is an integer to prioritize event filters: the greater the priority, the more it is prioritized. Negative integers are allowed.

The default priority is 0.

process_event()

process_event() = #ev_process{pid = pid(), change = {'DOWN', any()}}

An event record representing a monitored process that exited.

The event has the following fields:

pid: the PID of the terminated process
reason: the exit reason

process_event_filter()

process_event_filter() = #evf_process{pid = pid(), props = khepri_evf:process_event_filter_props()}

A process event filter.

It takes a PID to monitor and optionally properties.

process_event_filter_props()

process_event_filter_props() = #{priority => khepri_evf:priority()}

Tree event filter properties.

The properties are:

priority: a priority()

A Khepri path, whether it is a native path or a Unix-like path, can be used as a tree event filter. It will be automatically converted to a tree event filter with default properties.

tree_event()

tree_event() = #ev_tree{path = khepri_path:native_path(), change = khepri_tree:applied_change()}

An event record representing a change tree change.

The event has the following fields:

path: the path of the affected tree node
change: the nature of the change (create, update or delete).

tree_event_filter()

tree_event_filter() = #evf_tree{path = khepri_path:native_pattern(), props = khepri_evf:tree_event_filter_props()}

A tree event filter.

It takes a path pattern to monitor and optionally properties.

tree_event_filter_props()

tree_event_filter_props() = #{on_actions => [khepri_tree:change_type()], priority => khepri_evf:priority()}

Tree event filter properties.

The properties are:

on_actions: a list of actions to filter among create, update and delete; the default is to react to all of them.
priority: a priority()

A Khepri path, whether it is a native path or a Unix-like path, can be used as a tree event filter. It will be automatically converted to a tree event filter with default properties.

Function Index

tree/1	Constructs a tree event filter.
tree/2	Constructs a tree event filter.
process/1	Constructs a process event filter.
process/2	Constructs a process event filter.
wrap/1	Automatically detects the event filter type and ensures it is wrapped in one of the internal types.
get_priority/1	Returns the priority of the event filter.
set_priority/2	Sets the priority of the event filter.