Defines the behaviour an event handler must implement and provides a convenience macro that implements the behaviour, allowing you to handle only the events you are interested in processing.

You should start your event handlers using a Supervisor to ensure they are restarted on error.

Example

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler"

  def handle(%AnEvent{..}, _metadata) do
    # ... process the event
    :ok
  end
end

Start your event handler process (or use a Supervisor):

{:ok, _handler} = ExampleHandler.start_link()

Event handler name

The name you specify is used when subscribing to the event store. You must use a unique name for each event handler and process manager you start. Also, you should not change the name once the handler has been deployed. A new subscription will be created if you change the name and the event handler will receive already handled events.

You can use the module name of your event handler using the __MODULE__ special form:

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__
end

Subscription options

You can choose to start the event handler's event store subscription from :origin, :current position, or an exact event number using the start_from option. The default is to use the origin so your handler will receive all events.

Use the :current position when you don't want newly created event handlers to go through all previous events. An example would be adding an event handler to send transactional emails to an already deployed system containing many historical events.

The start_from option only applies when the subscription is initially created, the first time the handler starts. Whenever the handler restarts the subscription will resume from the next event after the last successfully processed event. Restarting an event handler does not restart its subscription.

Example

Set the start_from option (:origin, :current, or an explicit event number) when using Commanded.Event.Handler:

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler",
    start_from: :origin
end

You can optionally override :start_from by passing it as option when starting your handler:

{:ok, _handler} = ExampleHandler.start_link(start_from: :current)

Subscribing to an individual stream

By default event handlers will subscribe to all events appended to any stream. Provide a subscribe_to option to subscribe to a single stream.

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__,
    subscribe_to: "stream1234"
end

This will ensure the handler only receives events appended to that stream.

Runtime event handler configuration

Runtime options can be provided to the event handler's start_link/1 function or its child spec. The init/1 callback function can also be used to define runtime configuration.

Example

Provide runtime configuration to start_link/1:

{:ok, _pid} =
  ExampleHandler.start_link(
    application: ExampleApp,
    name: "ExampleHandler"
  )

Or when supervised:

Supervisor.start_link([
  {ExampleHandler, application: ExampleApp, name: "ExampleHandler"}
], strategy: :one_for_one)

Event handler state

An event handler can define and update state which is held in the GenServer process memory. It is passed to the handle/2 function as part of the metadata using the :state key. The state is transient and will be lost whenever the process restarts.

Initial state can be set in the init/1 callback function by adding a :state key to the config. It can also be provided when starting the handler process:

ExampleHandler.start_link(state: initial_state)

Or when supervised:

Supervisor.start_link([
  {ExampleHandler, state: initial_state}
], strategy: :one_for_one)

State can be updated by returning {:ok, new_state} from any handle/2 function. Returning an :ok reply will keep the state unchanged.

Handler state is also included in the Commanded.Event.FailureContext struct passed to the error/3 callback function.

Example

defmodule StatefulEventHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__

  def init(config) do
    config = Keyword.put_new(config, :state, %{})

    {:ok, config}
  end

  def handle(event, metadata) do
    %{state: state} = metadata

    new_state = mutate_state(state)

    {:ok, new_state}
  end
end

Concurrency

An event handler may be configured to start multiple processes to handle the events concurrently. By default one process will be started, processing events one at a time in order. The :concurrency option determines how many event handler processes are started. It must be a positive integer.

Note with concurrent processing events will likely by processed out of order. If you need to enforce an order, such as per stream or by using a field from an event, you can define a partition_by/2 callback function in the event handler module. The function will receive each event and its metadata and must return a consistent term indicating the event's partition. Events which return the same term are guaranteed to be processed in order by the same event handler instance. While events with different partitions may be processed concurrently by another instance. An attempt will be made to distribute events as evenly as possible to all running event handler instances.

Only :eventual consistency is supported when multiple handler processes are configured with a :concurrency of greater than one.

Example

defmodule ConcurrentProcssingEventHandler do
  alias Commanded.EventStore.RecordedEvent

  use Commanded.Event.Handler,
    application: ExampleApp,
    name: __MODULE__,
    concurrency: 10

  def init(config) do
    # Fetch the index of this event handler instance (0..9 in this example)
    index = Keyword.fetch!(config, :index)

    {:ok, config}
  end

  def handle(event, metadata) do
    :ok
  end

  # Partition events by their stream
  def partition_by(event, metadata) do
    %{stream_id: stream_id} = metadata

    stream_id
  end
end

Consistency

For each event handler you can define its consistency, as one of either :strong or :eventual.

This setting is used when dispatching commands and specifying the consistency option.

When you dispatch a command using :strong consistency, after successful command dispatch the process will block until all event handlers configured to use :strong consistency have processed the domain events created by the command. This is useful when you have a read model updated by an event handler that you wish to query for data affected by the command dispatch. With :strong consistency you are guaranteed that the read model will be up-to-date after the command has successfully dispatched. It can be safely queried for data updated by any of the events created by the command.

The default setting is :eventual consistency. Command dispatch will return immediately upon confirmation of event persistence, not waiting for any event handlers.

Note strong consistency does not imply a transaction covers the command dispatch and event handling. It only guarantees that the event handler will have processed all events produced by the command: if event handling fails the events will have still been persisted.

Example

Define an event handler with :strong consistency:

defmodule ExampleHandler do
  use Commanded.Event.Handler,
    application: ExampleApp,
    name: "ExampleHandler",
    consistency: :strong
end

Dynamic application

An event handler's application can be provided as an option to start_link/1. This can be used to start the same handler multiple times, each using a separate Commanded application and event store.

Example

Start an event handler process for each tenant in a multi-tenanted app, guaranteeing that the data and processing remains isolated between tenants.

for tenant <- [:tenant1, :tenant2, :tenant3] do
  {:ok, _app} = MyApp.Application.start_link(name: tenant)
  {:ok, _handler} = ExampleHandler.start_link(application: tenant)
end

Typically you would start the event handlers using a supervisor:

children =
  for tenant <- [:tenant1, :tenant2, :tenant3] do
    {ExampleHandler, application: tenant}
  end

Supervisor.start_link(children, strategy: :one_for_one)

The above example requires three named Commanded applications to have already been started.

Telemetry

• [:commanded, :event, :handle, :start]

  • Description: Emitted when an event handler starts handling an event
  • Measurements: %{system_time: integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), recorded_event: RecordedEvent.t()}

• [:commanded, :event, :handle, :stop]

  • Description: Emitted when an event handler stops handling an event
  • Measurements: %{duration: non_neg_integer()}
  • Metadata: %{:application => Commanded.Application.t(), :context => map(), :handler_name => String.t(), :handler_module => atom(), :handler_state => map(), :recorded_event => RecordedEvent.t(), optional(:error) => any()}

• [:commanded, :event, :handle, :exception]

  • Description: Emitted when an event handler raises an exception
  • Measurements: %{duration: non_neg_integer()}
  • Metadata: %{application: Commanded.Application.t(), context: map(), handler_name: String.t(), handler_module: atom(), handler_state: map(), recorded_event: RecordedEvent.t(), kind: :throw | :error | :exit, reason: any(), stacktrace: list()}